One post tagged with "opentelemetry"

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
    "go.opentelemetry.io/otel/propagation"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

func initTracer(ctx context.Context) (*sdktrace.TracerProvider, error) {
    exp, err := otlptracehttp.New(ctx)
    if err != nil {
        return nil, err
    }
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
        sdktrace.WithBatcher(exp),
    )
    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.TraceContext{})
    return tp, nil
}

// Sampler decides whether a trace should be sampled and exported.
type Sampler interface {
    // DO NOT CHANGE: any modification will not be backwards compatible and
    // must never be done outside of a new major release.

    // ShouldSample returns a SamplingResult based on a decision made from the
    // passed parameters.
    ShouldSample(parameters SamplingParameters) SamplingResult
    // DO NOT CHANGE: any modification will not be backwards compatible and
    // must never be done outside of a new major release.

    // Description returns information describing the Sampler.
    Description() string
    // DO NOT CHANGE: any modification will not be backwards compatible and
    // must never be done outside of a new major release.
}

import (
    "go.opentelemetry.io/otel"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

// kubegems sampler, ignore samples whitch contains "kubegems.ignore" attrbute.
type kubegemsSampler struct{}

func (as kubegemsSampler) ShouldSample(p sdktrace.SamplingParameters) sdktrace.SamplingResult {
    result := sdktrace.SamplingResult{
        Tracestate: trace.SpanContextFromContext(p.ParentContext).TraceState(),
    }
    shouldSample := true
    for _, att := range p.Attributes {
        if att.Key == "kubegems.ignore" && att.Value.AsBool() == true {
            shouldSample = false
            break
        }
    }
    if shouldSample {
        result.Decision = sdktrace.RecordAndSample
    } else {
        result.Decision = sdktrace.Drop
    }
    return result
}

func (as kubegemsSampler) Description() string {
    return "KubegemsSampler"
}

// get user name by user id
func getUser(ctx context.Context, id string) (string, error) {
    // start a new span from context.
    newCtx, span := tracer.Start(ctx, "getUser", trace.WithAttributes(attribute.String("user.id", id)))
    defer span.End()
    // add start event
    span.AddEvent("start to get user",
        trace.WithTimestamp(time.Now()),
    )
    var username string
    // get user name from db, if you want to trace it, `WithContext` is necessary.
    result := getDB().WithContext(newCtx).Raw(`select username from users where id = ?`, id).Scan(&username)
    if result.Error != nil || result.RowsAffected == 0 {
        err := fmt.Errorf("user %s not found", id)
        span.SetStatus(codes.Error, err.Error())
        return "", err
    }
    // set user info in span's attributes
    span.SetAttributes(attribute.String("user.name", username))
    // add end event
    span.AddEvent("end to get user",
        trace.WithTimestamp(time.Now()),
        trace.WithAttributes(attribute.String("user.name", username)),
    )
    span.SetStatus(codes.Ok, "")
    return username, nil
}

    otel.SetTextMapPropagator(propagation.TraceContext{})

import (
    "net/http"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
)

func DoRequest(){
    ...
    req, err := http.NewRequestWithContext(ctx, method, addr, body)
    // inject to http.Request by propagator to do distribute tracing
    otel.GetTextMapPropagator().Inject(req.Context(), propagation.HeaderCarrier(req.Header))
    http.DefaultClient.Do(req)
    ...   
}

import (
    "go.opentelemetry.io/otel/propagation"
)   

func HandleRequest(){
    ...
    // extract from http.Request by propagator to do distribute tracing
    ctx := cfg.Propagators.Extract(req.Context(), propagation.HeaderCarrier(req.Header))
    ctx, span := tracer.Start(ctx, spanName, opts...)
    defer span.End()
    req = req.WithContext(ctx)
    ...
}

    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))

import (
    "net/http"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/baggage"
)

func DoRequest(){
    ...
    userBaggage, err := baggage.Parse(fmt.Sprintf("user.id=%d,user.name=%s", user.ID, user.Username))
    if err != nil {
        otel.Handle(err)
    }

    req, err := http.NewRequestWithContext(baggage.ContextWithBaggage(ctx, userBaggage), clientreq.Method, addr, body)
    if err != nil {
        return nil, err
    }
    otel.GetTextMapPropagator().Inject(req.Context(), propagation.HeaderCarrier(req.Header))
    http.DefaultClient.Do(req)
    ...
}

import (
    "go.opentelemetry.io/otel/propagation"
        "go.opentelemetry.io/otel/baggage"
)   

func HandleRequest(){
    ...
    // extract from http.Request by propagator to do distribute tracing
    ctx := cfg.Propagators.Extract(req.Context(), propagation.HeaderCarrier(req.Header))
    ctx, span := tracer.Start(ctx, spanName, opts...)
    defer span.End()
    
    reqBaggage := baggage.FromContext(ctx)
    span.SetAttributes(
        attribute.String("user.id", reqBaggage.Member("user.id").Value()),
        attribute.String("user.name", reqBaggage.Member("user.name").Value()),
    )    
    req = req.WithContext(ctx)
    ...
}

// TextMapCarrier is the storage medium used by a TextMapPropagator.
type TextMapCarrier interface {
    ...
}

// HeaderCarrier adapts http.Header to satisfy the TextMapCarrier interface.
type HeaderCarrier http.Header

import (
    "context"
    "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
    "go.opentelemetry.io/otel/metric/global"
    sdkmetric "go.opentelemetry.io/otel/sdk/metric"
)

func initMeter(ctx context.Context) (*sdkmetric.MeterProvider, error) {
    exp, err := otlpmetrichttp.New(ctx)
    if err != nil {
        return nil, err
    }
    mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exp, sdkmetric.WithInterval(15*time.Second))))
    global.SetMeterProvider(mp)
    return mp, nil
}

import (
    "time"

    "github.com/gin-gonic/gin"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/metric/global"
    "go.opentelemetry.io/otel/metric/instrument/syncfloat64"
    "go.opentelemetry.io/otel/metric/instrument/syncint64"
    "go.opentelemetry.io/otel/propagation"
    semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
)

// Server HTTP metrics.
const (
    RequestCount          = "http.server.request_count"           // Incoming request count total
    ServerLatency         = "http.server.duration"                // Incoming end to end duration, microseconds
)

const (
    instrumentationName = "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

var (
    counters       map[string]syncint64.Counter
    valueRecorders map[string]syncfloat64.Histogram
)

func MeterMiddleware(service string) gin.HandlerFunc {
    counters = make(map[string]syncint64.Counter)
    valueRecorders = make(map[string]syncfloat64.Histogram)
    meter := global.MeterProvider().Meter(instrumentationName)

    requestCounter, _ := meter.SyncInt64().Counter(RequestCount)
    serverLatencyMeasure, _ := meter.SyncFloat64().Histogram(ServerLatency)

    counters[RequestCount] = requestCounter
    valueRecorders[ServerLatency] = serverLatencyMeasure
    return func(c *gin.Context) {
        requestStartTime := time.Now()
        attributes := semconv.HTTPServerMetricAttributesFromHTTPRequest(service, c.Request)
        ctx := otel.GetTextMapPropagator().Extract(c.Request.Context(), propagation.HeaderCarrier(c.Request.Header))

        c.Next()
        // Use floating point division here for higher precision (instead of Millisecond method).
        // 由于Bucket分辨率的问题，这里只能记录为millseconds而不是seconds
        elapsedTime := float64(time.Since(requestStartTime)) / float64(time.Millisecond)
        counters[RequestCount].Add(ctx, 1, attributes...)
        valueRecorders[ServerLatency].Record(ctx, elapsedTime, attributes...)
    }
}

func logMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        start := time.Now()
        ctx := otel.GetTextMapPropagator().Extract(c.Request.Context(), propagation.HeaderCarrier(c.Request.Header))
        span := trace.SpanFromContext(ctx)

        c.Next()
        statusCode := c.Writer.Status()
        logrus.WithFields(logrus.Fields{
            "method":   c.Request.Method,
            "path":     c.Request.URL.Path,
            "trace-id": span.SpanContext().TraceID(),
            "code":     statusCode,
            "latency":  time.Since(start).String(),
            "sampled":  span.SpanContext().IsSampled(),
        }).Info(http.StatusText(statusCode))
    }
}

    beego.InsertFilter("*", beego.BeforeRouter, func(c *bcontext.Context) {
        ctx := otel.GetTextMapPropagator().Extract(c.Request.Context(), propagation.HeaderCarrier(c.Request.Header))
        newctx, span := tracer.Start(ctx, "getUserFromBaggage")
        defer span.End()
        logrus.WithFields(logrus.Fields{
            "method":   c.Request.Method,
            "path":     c.Request.URL.Path,
            "trace-id": span.SpanContext().TraceID(),
            "sampled":  span.SpanContext().IsSampled(),
        }).Info("handle request")

        reqBaggage := baggage.FromContext(newctx)
        span.SetAttributes(
            attribute.String("user.id", reqBaggage.Member("user.id").Value()),
            attribute.String("user.name", reqBaggage.Member("user.name").Value()),
        )
        c.Request = c.Request.WithContext(newctx)
    })