How to configure pubsub topic, subscription and BigQuery for googlepubsub exporter for OTEL traces

[nabheet]

I have a working example of sending OTEL traces to BigQuery using pubsub but the data is stored in a binary column in a BQ Table. It also seems encoded in base64 when I query the data. I don't understand enough to know exactly how to accomplish this, but I need the data to be queryable in JSON format. I feel like it should be possible. I think these should be the steps:

1. Create schema based on proto buf.

syntax = "proto3";

package com.stuff;

message ExportTraceServiceRequest {
  message ResourceSpans {
    message Resource {
        string type = 1;
        map<string, string> labels = 2;
    }

    message Span {
        string traceId = 1;
        string spanId = 2;
        string name = 3;
        int64 startTime = 4;
        int64 endTime = 5;
        map<string, string> attributes = 6;
    }

    Resource resource = 1;
    repeated Span spans = 2;
  }

  repeated ResourceSpans resourceSpans = 1;
}

The messages must be nested otherwise the schema doesnt get created with too many objects or some error.

2. Create topic using above schema:

  schema = Schema(
        f"{topic_name}-schema",
        name=f"{topic_name}-schema",
        project=project_id,
        type="PROTOCOL_BUFFER",
        definition=topic_schema,
    )

    topic = Topic(
        topic_name,
        name=topic_name,
        project=project_id,
        schema_settings=TopicSchemaSettingsArgs(
            schema=schema,
            encoding="JSON",
        ),
    )

3. create table and subscription:

bq_table = Table(
        resource_name="otel-trace",
        table_id="otel_trace",
        friendly_name="otel_trace",
        project=project_id,
        dataset_id=big_query_dataset.dataset_id,
        time_partitioning=TableTimePartitioningArgs(
            type="DAY",
            field="publish_time",
        ),
    )

    subscription = Subscription(
        subscription_name,
        name=subscription_name,
        topic=topic.name,
        project=project_id,
        ack_deadline_seconds=20,
        bigquery_config=SubscriptionBigqueryConfigArgs(
            service_account_email=service_account_email,
            table=big_query_dataset.dataset_id.apply(
                lambda di: bq_table.table_id.apply(lambda ti: f"{project_id}:{di}.{ti}")
            ),
            use_topic_schema=True,
            use_table_schema=False,
            write_metadata=True,
            drop_unknown_fields=False,
        ),
        enable_message_ordering=True,
    )

Now I get the following error:

googleapi: Error 400: Field resourceSpans not found in table schema.
Details:
[
  {
    "@type": "type.googleapis.com/google.rpc.ErrorInfo",
    "domain": "pubsub.googleapis.com",
    "metadata": {
      "field_name": "resourceSpans",
      "reason": "NOT_FOUND",
      "source": "TABLE_SCHEMA"
    },
    "reason": "INCOMPATIBLE_SCHEMA"
  }
]

Does this mean that I still need to define a table schema even though I told the subscription to use the Topic schema? And what should it be?

I had this but I am quite confused so hoping someone who has done this before could guide me PLEASE!!!!!!

schema = [
        {"name": "attributes", "type": "JSON", "mode": "REQUIRED"},
        {"name": "data", "type": "BYTES", "mode": "REQUIRED"},
        {"name": "message_id", "type": "STRING", "mode": "REQUIRED"},
        {"name": "publish_time", "type": "TIMESTAMP", "mode": "REQUIRED"},
        {"name": "subscription_name", "type": "STRING", "mode": "REQUIRED"},
    ]

PS: the code examples are from Python Pulumi GCP.

[airhorns]

@nabheet did you end up figuring this out?

[nabheet]

No, I ended up implementing it a different way.

1. not using any schema in pubsub.
2. subscription is linked to a Cloud Function where I base64 decode it and format it to match my BQ table.