Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upgrade Opentelemetry Libraries and adopt enhancements (#2484)

Browse Source 
1. Use websocket support introduced in Openetelemetry Librarries
2. Use propagators settings introduced in Opentelemetry Go SDK
3. Use samplers settings introduced in Openetelemetry Go SDK

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2022-07-20 11:13:12 +05:30
committed by GitHub
parent dba9143f45
commit 5c886e4cf4
6 changed files with 81 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/router/functionHandler.go
+9 -27
View File
@@ -39,7 +39,6 @@ import (
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/error/network"
executorClient "github.com/fission/fission/pkg/executor/client"
"github.com/fission/fission/pkg/router/util"
"github.com/fission/fission/pkg/throttler"
"github.com/fission/fission/pkg/utils"
otelUtils "github.com/fission/fission/pkg/utils/otel"
@@ -318,39 +317,22 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re
dumpReqFunc(newReq)
}
// The otelhttp.NewTransport() does not work with WebSocket.
// This is probably because it modifies the response body.
// Until we find a better solution to handle websocket requests, we will continue to
// use http.Transport(). We check if the request isWebsocketRequest() and use the
// http.Transport() irrespective of open telemetry is enabled or not.
// Related issue: https://github.com/open-telemetry/opentelemetry-js-contrib/issues/12
// forward the request to the function service
var resp *http.Response
if util.IsWebsocketRequest(newReq) {
resp, err = transport.RoundTrip(newReq)
} else {
otelUtils.SpanTrackEvent(ctx, "roundtrip", otelUtils.MapToAttributes(map[string]string{
"function-name": fnMeta.Name,
"function-namespace": fnMeta.Namespace,
"function-url": newReq.URL.String(),
"retryCounter": fmt.Sprintf("%d", retryCounter)})...)
otelRoundTripper := otelhttp.NewTransport(transport)
resp, err = otelRoundTripper.RoundTrip(newReq)
otelUtils.SpanTrackEvent(ctx, "roundtrip", otelUtils.MapToAttributes(map[string]string{
"function-name": fnMeta.Name,
"function-namespace": fnMeta.Namespace,
"function-url": newReq.URL.String(),
"retryCounter": fmt.Sprintf("%d", retryCounter)})...)
otelRoundTripper := otelhttp.NewTransport(transport)
resp, err := otelRoundTripper.RoundTrip(newReq)
if roundTripper.funcHandler.isDebugEnv {
dumpRespFunc(resp)
}
if err == nil {
// return response back to user
if roundTripper.funcHandler.isDebugEnv {
dumpRespFunc(resp)
}
return resp, nil
}
if roundTripper.funcHandler.isDebugEnv && resp != nil {
dumpRespFunc(resp)
}
roundTripper.totalRetry++
if i >= roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1 {
pkg/utils/otel/provider.go
+6 -124
View File
@@ -2,19 +2,14 @@ package otel
import (
"context"
"fmt"
"os"
"strconv"
"strings"
"go.opentelemetry.io/contrib/propagators/aws/xray"
"go.opentelemetry.io/contrib/propagators/b3"
"go.opentelemetry.io/contrib/propagators/jaeger"
"go.opentelemetry.io/contrib/propagators/ot"
"go.opentelemetry.io/contrib/propagators/autoprop"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
@@ -25,12 +20,10 @@ import (
)
const (
OtelEnvPrefix = "OTEL_"
OtelEndpointEnvVar = "OTEL_EXPORTER_OTLP_ENDPOINT"
OtelInsecureEnvVar = "OTEL_EXPORTER_OTLP_INSECURE"
OtelTracesSampler = "OTEL_TRACES_SAMPLER"
OtelTracesSamplerArg = "OTEL_TRACES_SAMPLER_ARG"
OtelPropogaters = "OTEL_PROPOGATORS"
OtelEnvPrefix = "OTEL_"
OtelEndpointEnvVar = "OTEL_EXPORTER_OTLP_ENDPOINT"
OtelInsecureEnvVar = "OTEL_EXPORTER_OTLP_INSECURE"
OtelPropagaters = "OTEL_PROPAGATORS"
)
type OtelConfig struct {
@@ -38,111 +31,6 @@ type OtelConfig struct {
insecure bool
}
/*
Each Sampler type defines its own expected input, if any.
Currently we get trace ratio for the case of,
1. traceidratio
2. parentbased_traceidratio
*/
func getSamplerArg() (float64, error) {
arg := os.Getenv(OtelTracesSamplerArg)
return strconv.ParseFloat(arg, 64)
}
/* GetPropogater returns a slice of propagators to be used by the OpenTelemetry
provider.
Supported providers:
tracecontext - W3C Trace Context
baggage - W3C Baggage
b3 - B3 Single
b3multi - B3 Multi
jaeger - Jaeger uber-trace-id header
xray - AWS X-Ray (third party)
ottrace - OpenTracing Trace (third party)
*/
func GetPropogater(logger *zap.Logger) []propagation.TextMapPropagator {
propogatersEnv := os.Getenv(OtelPropogaters)
if propogatersEnv == "" {
return []propagation.TextMapPropagator{
propagation.TraceContext{}, propagation.Baggage{},
}
}
propogators := []propagation.TextMapPropagator{}
for _, prop := range strings.Split(propogatersEnv, ",") {
switch prop {
case "tracecontext":
propogators = append(propogators, propagation.TraceContext{})
case "baggage":
propogators = append(propogators, propagation.Baggage{})
case "b3multi":
propogators = append(propogators, b3.New(b3.WithInjectEncoding(b3.B3MultipleHeader)))
case "b3":
propogators = append(propogators, b3.New(b3.WithInjectEncoding(b3.B3SingleHeader)))
case "jaeger":
propogators = append(propogators, jaeger.Jaeger{})
case "xray":
propogators = append(propogators, xray.Propagator{})
case "ottrace":
propogators = append(propogators, ot.OT{})
default:
logger.Error("Unsupported propagation type", zap.String("propagation", prop))
}
}
if len(propogators) == 0 {
return []propagation.TextMapPropagator{
propagation.TraceContext{}, propagation.Baggage{},
}
}
return propogators
}
/*
GetSampler returns a sampler that can be used to sample traces.
This is based on https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/sdk-environment-variables.md#general-sdk-configuration
We have to implement as open-telemetry Go sdk doesn't support configuration of different samplers.
Once its added we may remove this code.
Supported samplers:
always_on - Sampler that always samples spans, regardless of the parent span's sampling decision.
always_off - Sampler that never samples spans, regardless of the parent span's sampling decision.
traceidratio - Sampler that samples probabalistically based on rate.
parentbased_always_on - (default) Sampler that respects its parent span's sampling decision, but otherwise always samples.
parentbased_always_off - Sampler that respects its parent span's sampling decision, but otherwise never samples.
parentbased_traceidratio - Sampler that respects its parent span's sampling decision, but otherwise samples probabalistically based on rate.
Environment variables:
OTEL_TRACES_SAMPLER - Sampler to use(one of the above samplers)
OTEL_TRACES_SAMPLER_ARG - Argument to pass to the sampler(float value)
*/
func GetSampler() (sdktrace.Sampler, error) {
samplerType := os.Getenv(OtelTracesSampler)
switch samplerType {
case "always_on":
return sdktrace.AlwaysSample(), nil
case "always_off":
return sdktrace.NeverSample(), nil
case "parentbased_always_on":
return sdktrace.ParentBased(sdktrace.AlwaysSample()), nil
case "parentbased_always_off":
return sdktrace.ParentBased(sdktrace.NeverSample()), nil
case "traceidratio":
arg, err := getSamplerArg()
if err != nil {
return nil, fmt.Errorf("invalid sampler arg: %w", err)
}
return sdktrace.TraceIDRatioBased(arg), nil
case "parentbased_traceidratio":
arg, err := getSamplerArg()
if err != nil {
return nil, fmt.Errorf("invalid sampler arg: %w", err)
}
return sdktrace.ParentBased(sdktrace.TraceIDRatioBased(arg)), nil
default:
return sdktrace.ParentBased(sdktrace.AlwaysSample()), nil
}
}
// parseOtelConfig parses the environment variables OTEL_EXPORTER_OTLP_ENDPOINT and
func parseOtelConfig() OtelConfig {
config := OtelConfig{}
@@ -191,13 +79,8 @@ func InitProvider(ctx context.Context, logger *zap.Logger, serviceName string) (
if err != nil {
return nil, err
}
sampler, err := GetSampler()
if err != nil {
return nil, err
}
tracerProvider := sdktrace.NewTracerProvider(
sdktrace.WithResource(res),
sdktrace.WithSampler(sampler),
)
traceExporter, err := getTraceExporter(ctx, logger)
if err != nil {
@@ -210,8 +93,7 @@ func InitProvider(ctx context.Context, logger *zap.Logger, serviceName string) (
}
otel.SetTracerProvider(tracerProvider)
propogaters := GetPropogater(logger)
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propogaters...))
otel.SetTextMapPropagator(autoprop.NewTextMapPropagator())
// Shutdown will flush any remaining spans and shut down the exporter.
return func(ctx context.Context) {
err := tracerProvider.Shutdown(ctx)
pkg/utils/otel/provider_test.go
+17 -108
View File
@@ -2,145 +2,54 @@ package otel
import (
"context"
"errors"
"fmt"
"os"
"reflect"
"sort"
"testing"
"go.opentelemetry.io/contrib/propagators/jaeger"
"go.opentelemetry.io/contrib/propagators/autoprop"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
"go.opentelemetry.io/otel/propagation"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"github.com/fission/fission/pkg/utils/loggerfactory"
)
func TestGetPropogater(t *testing.T) {
if OtelPropogaters != "OTEL_PROPOGATORS" {
t.Errorf("Expected OTEL_PROPOGATORS to be set, got %s", OtelPropogaters)
if OtelPropagaters != "OTEL_PROPAGATORS" {
t.Errorf("Expected OTEL_PROPOGATORS to be set, got %s", OtelPropagaters)
}
// tracecontext, baggage, b3, b3multi, jaeger, xray, ottrace, and none
tests := []struct {
propogaterEnv string
propogaters []propagation.TextMapPropagator
propogaters []string
}{
{
"",
[]propagation.TextMapPropagator{propagation.TraceContext{}, propagation.Baggage{}},
"none",
[]string{},
},
{
"tracecontext,baggage",
[]propagation.TextMapPropagator{propagation.TraceContext{}, propagation.Baggage{}},
[]string{"baggage", "traceparent", "tracestate"},
},
{
"jaeger",
[]propagation.TextMapPropagator{jaeger.Jaeger{}},
[]string{"uber-trace-id"},
},
{
"baggage,tracecontext",
[]propagation.TextMapPropagator{propagation.Baggage{}, propagation.TraceContext{}},
[]string{"baggage", "traceparent", "tracestate"},
},
{
"jaeger,baggage",
[]propagation.TextMapPropagator{jaeger.Jaeger{}, propagation.Baggage{}},
},
}
logger := loggerfactory.GetLogger()
for _, tt := range tests {
os.Setenv(OtelPropogaters, tt.propogaterEnv)
prop := GetPropogater(logger)
if prop == nil {
t.Errorf("GetPropogater() = %#v, want %#v", prop, tt.propogaters)
}
if len(prop) != len(tt.propogaters) {
t.Errorf("GetPropogater() = %#v, want %#v", prop, tt.propogaters)
}
if !reflect.DeepEqual(prop, tt.propogaters) {
t.Errorf("GetPropogater() = %#v, want %#v", prop, tt.propogaters)
}
}
}
func TestGetSampler(t *testing.T) {
if OtelTracesSampler != "OTEL_TRACES_SAMPLER" {
t.Errorf("Expected OTEL_TRACES_SAMPLER to be set, got %s", OtelTracesSampler)
}
if OtelTracesSamplerArg != "OTEL_TRACES_SAMPLER_ARG" {
t.Errorf("Expected OTEL_TRACES_SAMPLER_ARG to be set, got %s", OtelTracesSamplerArg)
}
if OtelPropogaters != "OTEL_PROPOGATORS" {
t.Errorf("Expected OTEL_PROPOGATORS to be set, got %s", OtelPropogaters)
}
tests := []struct {
sampler string
samplerArg string
wantSampler sdktrace.Sampler
wantError error
}{
{
"",
"",
sdktrace.ParentBased(sdktrace.AlwaysSample()),
nil,
},
{
"always_on",
"",
sdktrace.AlwaysSample(),
nil,
},
{
"always_off",
"",
sdktrace.NeverSample(),
nil,
},
{
"parentbased_always_on",
"",
sdktrace.ParentBased(sdktrace.AlwaysSample()),
nil,
},
{
"parentbased_always_off",
"",
sdktrace.ParentBased(sdktrace.NeverSample()),
nil,
},
{
"traceidratio",
"0.5",
sdktrace.TraceIDRatioBased(0.5),
nil,
},
{
"traceidratio",
"",
nil,
errors.New("invalid sampler arg: strconv.ParseFloat: parsing \"\": invalid syntax"),
},
{
"parentbased_traceidratio",
"",
nil,
errors.New("invalid sampler arg: strconv.ParseFloat: parsing \"\": invalid syntax"),
},
{
"parentbased_traceidratio",
"0.01",
sdktrace.ParentBased(sdktrace.TraceIDRatioBased(0.01)),
nil,
[]string{"baggage", "uber-trace-id"},
},
}
for _, tt := range tests {
os.Setenv(OtelTracesSampler, tt.sampler)
os.Setenv(OtelTracesSamplerArg, tt.samplerArg)
gotSampler, gotError := GetSampler()
if !reflect.DeepEqual(gotSampler, tt.wantSampler) {
t.Errorf("GetSampler() gotSampler = %#v, want %#v", gotSampler, tt.wantSampler)
}
if fmt.Sprintf("%s", gotError) != fmt.Sprintf("%s", tt.wantError) {
t.Errorf("GetSampler() gotError = %#v, want %#v", gotError, tt.wantError)
os.Setenv(OtelPropagaters, tt.propogaterEnv)
propFields := autoprop.NewTextMapPropagator().Fields()
sort.Strings(propFields)
if !reflect.DeepEqual(propFields, tt.propogaters) {
t.Errorf("Expected %s, got %s", tt.propogaters, propFields)
}
}
}