frx/kurious
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

support metrics sending via grpc

Browse Source
This commit is contained in:
Aleksandr Trushkin
2024-09-25 00:47:12 +03:00
parent c7fada2c54
commit c3d6510a7d
6 changed files with 179 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
cmd/kuriweb/http.go
View File

@ -16,7 +16,8 @@ import (
"github.com/gorilla/mux"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
"go.opentelemetry.io/otel/metric"
semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
"go.opentelemetry.io/otel/trace"
)
@ -57,6 +58,7 @@ func setupHTTP(cfg config.HTTP, srv xhttp.Server, log *slog.Logger) *http.Server
mux.CORSMethodMiddleware(router),
middlewareLogger(log),
middlewareTrace(),
middlewareMetrics(),
)
setupCoursesHTTP(srv, router, log)
@ -114,6 +116,53 @@ func (k attributeStringKey) Value(value string) attribute.KeyValue {
return attribute.String(string(k), value)
}
func must[T any](value T, err error) T {
if err != nil {
panic(err.Error())
}
return value
}
func middlewareMetrics() mux.MiddlewareFunc {
requestDuration := must(webmetric.Float64Histogram(
semconv.HTTPServerRequestDurationName,
metric.WithUnit(semconv.HTTPServerRequestDurationUnit),
metric.WithDescription(semconv.HTTPServerRequestDurationDescription),
))
f := func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
route := mux.CurrentRoute(r)
hpath, _ := route.GetPathTemplate()
attributes := make([]attribute.KeyValue, 0, 8)
attributes = append(
attributes,
semconv.HTTPRequestMethodOriginal(r.Method),
semconv.URLFull(hpath),
semconv.URLPath(r.URL.Path),
)
next.ServeHTTP(w, r)
if wr, ok := w.(*customResponseWriter); ok {
statusCode := xdefault.WithFallback(wr.statusCode, http.StatusOK)
attributes = append(attributes,
semconv.HTTPResponseStatusCode(statusCode),
semconv.HTTPResponseBodySize(wr.wroteBytes),
)
}
elapsed := time.Since(start).Truncate(time.Millisecond)
requestDuration.Record(
r.Context(), elapsed.Seconds(), metric.WithAttributes(attributes...))
})
}
return f
}
func middlewareTrace() mux.MiddlewareFunc {
reqidAttr := attributeStringKey("http.request_id")

106
cmd/kuriweb/trace.go
View File

@ -8,9 +8,11 @@ import (
"time"
"git.loyso.art/frx/kurious/internal/common/config"
"google.golang.org/grpc/encoding/gzip"
"github.com/gorilla/mux"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
@ -18,12 +20,16 @@ import (
"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/metric"
"go.opentelemetry.io/otel/sdk/metric/metricdata"
"go.opentelemetry.io/otel/sdk/resource"
"go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
)
var webtracer = otel.Tracer("kuriweb.http")
var (
webtracer = otel.Tracer("kuriweb.http")
webmetric = otel.Meter("kuriweb.http")
)
type shutdownFunc func(context.Context) error
@ -44,6 +50,11 @@ func setupOtelSDK(ctx context.Context, cfg config.Trace) (shutdown shutdownFunc,
return err
}
resource, err := makeServiceResource(ctx)
if err != nil {
return shutdown, fmt.Errorf("making service resource: %w", err)
}
prop := newPropagator()
otel.SetTextMapPropagator(prop)
@ -52,7 +63,7 @@ func setupOtelSDK(ctx context.Context, cfg config.Trace) (shutdown shutdownFunc,
Type: cfg.Type,
AuthHeader: cfg.APIHeader,
APIKey: cfg.APIKey,
})
}, resource)
if err != nil {
return nil, handleError(err)
}
@ -60,7 +71,13 @@ func setupOtelSDK(ctx context.Context, cfg config.Trace) (shutdown shutdownFunc,
otel.SetTracerProvider(tracerProvider)
if cfg.ShowMetrics {
meterProvider, err := newMeterProvider()
meterProvider, err := newCommonMeterProvider(ctx, meterProviderParams{
Endpoint: cfg.Endpoint,
Type: cfg.Type,
AuthHeaderKey: cfg.APIHeader,
AuthHeaderValue: cfg.APIKey,
ReadInterval: time.Second * 15,
}, resource)
if err != nil {
return nil, handleError(err)
}
@ -79,13 +96,13 @@ func newPropagator() propagation.TextMapPropagator {
}
type TraceProviderParams struct {
Type config.TraceClientType
Endpoint string
APIKey string
AuthHeader string
Type config.TraceClientType
}
func newCommonTraceProvider(ctx context.Context, params TraceProviderParams) (tp *trace.TracerProvider, err error) {
func makeServiceResource(ctx context.Context) (*resource.Resource, error) {
r, err := resource.New(
ctx,
resource.WithDetectors(
@ -101,12 +118,10 @@ func newCommonTraceProvider(ctx context.Context, params TraceProviderParams) (tp
if err != nil {
return nil, fmt.Errorf("making new resource: %w", err)
}
return resource.Merge(resource.Default(), r)
}
r, err = resource.Merge(resource.Default(), r)
if err != nil {
return nil, fmt.Errorf("merging resources: %w", err)
}
func newCommonTraceProvider(ctx context.Context, params TraceProviderParams, r *resource.Resource) (tp *trace.TracerProvider, err error) {
opts := make([]trace.TracerProviderOption, 0, 4)
opts = append(
opts,
@ -130,7 +145,7 @@ func newCommonTraceProvider(ctx context.Context, params TraceProviderParams) (tp
otlptracegrpc.WithEndpointURL(params.Endpoint),
otlptracegrpc.WithInsecure(),
otlptracegrpc.WithHeaders(headers),
otlptracegrpc.WithCompressor("gzip"),
otlptracegrpc.WithCompressor(gzip.Name),
)
case config.TraceClientTypeHTTP:
httpClient := otlptracehttp.NewClient(
@ -158,21 +173,70 @@ func newCommonTraceProvider(ctx context.Context, params TraceProviderParams) (tp
return tp, nil
}
func newMeterProvider() (*metric.MeterProvider, error) {
metricExporter, err := stdoutmetric.New()
if err != nil {
return nil, err
type meterProviderParams struct {
Endpoint string
AuthHeaderKey string
AuthHeaderValue string
ReadInterval time.Duration
Type config.TraceClientType
}
func newCommonMeterProvider(ctx context.Context, params meterProviderParams, r *resource.Resource) (*metric.MeterProvider, error) {
var exporter metric.Exporter
var err error
switch params.Type {
case config.TraceClientTypeGRPC:
headers := make(map[string]string, 1)
if params.AuthHeaderKey != "" {
headers[params.AuthHeaderKey] = params.AuthHeaderValue
}
exporter, err = otlpmetricgrpc.New(
ctx,
otlpmetricgrpc.WithEndpointURL(params.Endpoint),
otlpmetricgrpc.WithHeaders(headers),
otlpmetricgrpc.WithCompressor(gzip.Name),
otlpmetricgrpc.WithTemporalitySelector(preferDeltaTemporalitySelector),
otlpmetricgrpc.WithInsecure(),
)
if err != nil {
return nil, fmt.Errorf("making grpc exporter: %w", err)
}
case config.TraceClientTypeStdout:
exporter, err = stdoutmetric.New()
if err != nil {
return nil, fmt.Errorf("making stdout exporter: %w", err)
}
default:
return nil, nil
}
meterProvider := metric.NewMeterProvider(
metric.WithReader(metric.NewPeriodicReader(metricExporter,
// Default is 1m. Set to 3s for demonstrative purposes.
metric.WithInterval(60*time.Second))),
reader := metric.NewPeriodicReader(
exporter,
metric.WithInterval(params.ReadInterval),
)
return meterProvider, nil
provider := metric.NewMeterProvider(
metric.WithReader(reader),
metric.WithResource(r),
)
return provider, nil
}
func muxHandleFunc(router *mux.Router, name, path string, hf http.HandlerFunc) *mux.Route {
// h := otelhttp.WithRouteTag(path, hf)
return router.Handle(path, hf).Name(name)
}
func preferDeltaTemporalitySelector(kind metric.InstrumentKind) metricdata.Temporality {
switch kind {
case metric.InstrumentKindCounter,
metric.InstrumentKindObservableCounter,
metric.InstrumentKindHistogram:
return metricdata.DeltaTemporality
default:
return metricdata.CumulativeTemporality
}
}