Integrate on Google Cloud Functions GoLang

When your Cloud Function has an HTTP trigger, you can instrument it with the help of the otelhttp package. First, install the package:

shellcopydownload
go get go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp

For ease of use, you can create a wrapper function for your entry point. The following code samples contain several utility functions which will help you set the required attributes on your spans.

gocopydownload
package instrumentation

import (
    "net/http"

    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
    semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
    "go.opentelemetry.io/otel/trace"

    "cloud.google.com/go/compute/metadata"

    dtTrace "github.com/dynatrace-oss/opentelemetry-exporter-go/core/trace"
)

type HttpHandler = func(w http.ResponseWriter, r *http.Request)

func InstrumentHandler(
    functionName string,
    function HttpHandler,
    tracerProvider *dtTrace.DtTracerProvider) HttpHandler {

    // get project id and region from Google Compute Engine metadata
    projectId, err := getProjectId()
    if err != nil {
        // handle error
    }
    region, err := getRegion()
    if err != nil {
        // handle error
    }

    // set at least these required attributes
    opts := []trace.SpanStartOption{
        trace.WithAttributes(
            semconv.FaaSTriggerHTTP,
            semconv.CloudProviderGCP,
            semconv.CloudPlatformGCPCloudFunctions,
            semconv.FaaSIDKey.String(createFaasId(projectId, region, functionName)),
            semconv.FaaSNameKey.String(functionName),
            semconv.CloudRegionKey.String(region),
        ),
    }

    // create instrumented handler
    handler := otelhttp.NewHandler(
        http.HandlerFunc(function), functionName, otelhttp.WithSpanOptions(opts...),
    )

    return func(w http.ResponseWriter, r *http.Request) {
        // call your function handler
        handler.ServeHTTP(w, r)

        // flush spans
        tracerProvider.ForceFlush(r.Context())
    }
}

func getProjectId() (string, error) {
    return metadata.ProjectID()
}

func getRegion() (string, error) {
    // Returned string has the format "projects/<numeric-project-id>/regions/<region>"
    fullRegion, err := metadata.Get("instance/region")
    if err != nil {
        return "", err
    }
    fullRegionSplit := strings.Split(fullRegion, "/")
    region := fullRegionSplit[len(fullRegionSplit)-1]
    return region, nil
}

func createFaasId(projectId, region, functionName string) string {
    return fmt.Sprintf("//cloudfunctions.googleapis.com/projects/%s/locations/%s/functions/%s",
        projectId, region, functionName)
}

To put everything together, instrument your Google Cloud Function entry point as follows:

gocopydownload
package myfunction

import (
    "net/http"

    "instrumentation"
    "otelsetup"

    "github.com/GoogleCloudPlatform/functions-framework-go/functions"
)

func init() {
    // see https://cloud.google.com/functions/docs/configuring/env-var#runtime_environment_variables_set_automatically
    functionName, ok := os.LookupEnv("K_SERVICE")
    if !ok {
        // function name not found; assign a default or treat as an error
    }
    if tracerProvider, err := otelsetup.InitializeTracing(functionName); err == nil {
        instrumentedHandler := instrumentation.InstrumentHandler(functionName, handler, tracerProvider)
        functions.HTTP(functionName, instrumentedHandler)
    } else {
        // handle error, or register function anyway without instrumentation
    }
}

func handler(w http.ResponseWriter, r *http.Request) {
    // your code goes here
}

gocopydownload
func handler(w http.ResponseWriter, r *http.Request) {
    // your code goes here

    // content or a status code must be written to the ResponseWriter
    w.WriteHeader(http.StatusOK)
}

OpenTelemetry currently doesn't provide out-of-the-box instrumentation for Pub/Sub, so instrumenting a Pub/Sub triggered Cloud Function requires some manual configuration.

gocopydownload
package instrumentation

import (
    "context"
    "fmt"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/propagation"
    semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
    "go.opentelemetry.io/otel/trace"

    cloudevents "github.com/cloudevents/sdk-go/v2/event"
)

const (
    pubSubHandlerInstrName = "my.company.com/my-pubsub-handler-instrumentation-name"
    pubSubHandlerInstrVer  = "0.1.0"
)

type PubSubHandler = func(context.Context, shared.PubSubMessage) error
type CloudEventHandler = func(context.Context, cloudevents.Event) error

type Flusher interface {
    ForceFlush(context.Context) error
}

// cloudevents.Event.Data() will contain a JSON object "message" containing
// an object that can be serialized to shared.PubSubMessage
// https://github.com/googleapis/google-cloudevents/blob/main/proto/google/events/cloud/pubsub/v1/data.proto
type MessagePublishedData struct {
    Message PubSubMessage `json:"message"`
}

// https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
type PubSubMessage struct {
    Data        []byte            `json:"data"`
    Attributes  map[string]string `json:"attributes"`
    MessageId   string            `json:"messageId"`
    PublishTime string            `json:"publishTime"`
    OrderingKey string            `json:"orderingKey"`
}

func InstrumentTopicHandler(
    functionName,
    topicID string,
    function PubSubHandler,
    flush Flusher) CloudEventHandler {
    // wrap the PubSubHandler into an instrumented CloudEventHandler
    return func(ctx context.Context, e cloudevents.Event) error {
        // extract PubSubMessage from event
        msg, err := extractMessageFromEvent(e)
        if err != nil {
            return err
        }

        // begin instrumentation
        ctx, span := beforePubSubHandlerInvoke(functionName, ctx, topicID, msg)
        defer span.End()

        // invoke handler function
        err = function(ctx, msg)

        // after instrumentation
        afterPubSubHandlerInvoke(span, err)

        flush.ForceFlush(ctx)

        return err
    }
}

func extractMessageFromEvent(e cloudevents.Event) (PubSubMessage, error) {
    var msgPublishedData MessagePublishedData
    if err := e.DataAs(&msgPublishedData); err != nil {
        return PubSubMessage{}, fmt.Errorf("event.DataAs: %v", err)
    }
    return msgPublishedData.Message, nil
}

func beforePubSubHandlerInvoke(
    functionName string,
    ctx context.Context,
    topicID string,
    msg shared.PubSubMessage) (context.Context, trace.Span) {
    // extract attributes from the PubSubMessage into a context
    if msg.Attributes != nil {
        propagator := otel.GetTextMapPropagator()
        ctx = propagator.Extract(ctx, propagation.MapCarrier(msg.Attributes))
    }

    projectId, err := getProjectId()
    if err != nil {
        // handle error
    }
    region, err := getRegion()
    if err != nil {
        // handle error
    }

    opts := []trace.SpanStartOption{
       trace.WithSpanKind(trace.SpanKindConsumer),
       trace.WithAttributes(
          semconv.FaaSTriggerPubsub,
          semconv.CloudProviderGCP,
          semconv.CloudPlatformGCPCloudFunctions,
          semconv.MessagingSystemKey.String("gcp_pubsub"),
          semconv.MessagingDestinationKey.String(topicID),
          semconv.MessagingDestinationKindTopic,
          semconv.MessagingOperationProcess,
          semconv.MessagingMessageIDKey.String(msg.MessageId),
          semconv.FaaSIDKey.String(createFaasId(projectId, region, functionName)),
          semconv.FaaSNameKey.String(functionName),
          semconv.CloudRegionKey.String(region),
      ),
    }

    tracer := otel.GetTracerProvider().Tracer(
        pubSubHandlerInstrName,
        trace.WithInstrumentationVersion(pubSubHandlerInstrVer),
    )

    return tracer.Start(ctx, fmt.Sprintf("%s process", topicID), opts...)
}

func afterPubSubHandlerInvoke(span trace.Span, err error) {
    if err != nil {
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
    }
}

func getProjectId() (string, error) {
    return metadata.ProjectID()
}

func getRegion() (string, error) {
    // Returned string has the format "projects/<numeric-project-id>/regions/<region>"
    fullRegion, err := metadata.Get("instance/region")
    if err != nil {
        return "", err
    }
    fullRegionSplit := strings.Split(fullRegion, "/")
    region := fullRegionSplit[len(fullRegionSplit)-1]
    return region, nil
}

func createFaasId(projectId, region, functionName string) string {
    return fmt.Sprintf("//cloudfunctions.googleapis.com/projects/%s/locations/%s/functions/%s",
        projectId, region, functionName)
}

Then, instrument your Google Cloud Function entry point as follows:

gocopydownload
package myfunction

import (
    "context"
    "log"

    "instrumentation"
    "otelsetup"

    "github.com/GoogleCloudPlatform/functions-framework-go/functions"
)

func init() {
    // see https://cloud.google.com/functions/docs/configuring/env-var#runtime_environment_variables_set_automatically
    functionName, ok := os.LookupEnv("K_SERVICE")
    if !ok {
        // function name not found; assign a default or treat as an error
    }
    if tracerProvider, err := otelsetup.InitializeTracing(functionName); err == nil {
        const topicID = "myexampletopic"
        instrumentedHandler := instrumentation.InstrumentTopicHandler(functionName,
            topicID, handler, tracerProvider)
        functions.CloudEvent(functionName, instrumentedHandler)
    } else {
        // handle error, or register function anyway without instrumentation
    }
}

func handler(ctx context.Context, msg instrumentation.PubSubMessage) error {
    // your code goes here, e.g. extract data from the msg.Data payload
    return nil
}

Note: The code samples above apply to CloudEvent functions. The code would require minor adaptations, such as removing the MessagePublishedData object and unmarshalling the event data directly to a PubSubMessage object, when using background functions.