rogee/atomctl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: issues

Browse Source
This commit is contained in:
Rogee
2025-01-07 17:16:15 +08:00
parent 49f68ca533
commit c5d0b643cf
1 changed files with 141 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
templates/project/providers/otel/provider.go.tpl
View File

@@ -10,10 +10,13 @@ import (
"github.com/pkg/errors"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
"go.opentelemetry.io/otel/propagation"
sdkmetric "go.opentelemetry.io/otel/sdk/metric"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.15.0"
@@ -30,125 +33,192 @@ func Provide(opts ...opt.Option) error {
return container.Container.Provide(func(ctx context.Context) (*OTEL, error) {
o := &OTEL{
Tracer: otel.Tracer(config.ServiceName),
config: &config,
}
var err error
o.Resource, err = initResource(ctx, &config)
err := o.initResource(ctx)
if err != nil {
return o, errors.Wrapf(err, "Failed to create OpenTelemetry resource")
}
if config.EndpointHTTP != "" {
o.Exporter, o.SpanProcessor, err = initHTTPExporterAndSpanProcessor(ctx, &config)
if err != nil {
return o, errors.Wrapf(err, "Failed to create OpenTelemetry trace exporter")
}
} else if config.EndpointGRPC != "" {
o.Exporter, o.SpanProcessor, err = initGrpcExporterAndSpanProcessor(ctx, &config)
if err != nil {
return o, errors.Wrapf(err, "Failed to create OpenTelemetry trace exporter")
}
} else {
return o, errors.New("http or grpc endpoint is required")
if err := o.initMeterProvider(ctx); err != nil {
return o, errors.Wrapf(err, "Failed to create OpenTelemetry metric provider")
}
traceProvider := sdktrace.NewTracerProvider(
sdktrace.WithSampler(sdktrace.AlwaysSample()),
sdktrace.WithResource(o.Resource),
sdktrace.WithSpanProcessor(o.SpanProcessor),
)
otel.SetTracerProvider(traceProvider)
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
propagation.TraceContext{},
propagation.Baggage{},
))
container.AddCloseAble(func() {
cxt, cancel := context.WithTimeout(ctx, time.Second)
defer cancel()
if err := o.Exporter.Shutdown(cxt); err != nil {
otel.Handle(err)
}
})
if err := o.initTracerProvider(ctx); err != nil {
return o, errors.Wrapf(err, "Failed to create OpenTelemetry tracer provider")
}
return o, nil
}, o.DiOptions()...)
}
type OTEL struct {
Tracer trace.Tracer
Resource *resource.Resource
Exporter *otlptrace.Exporter
SpanProcessor sdktrace.SpanProcessor
config *Config
Tracer trace.Tracer
Resource *resource.Resource
}
func initResource(ctx context.Context, conf *Config) (*resource.Resource, error) {
func (o *OTEL) initResource(ctx context.Context) (err error) {
hostName, _ := os.Hostname()
r, err := resource.New(
o.Resource, err = resource.New(
ctx,
resource.WithFromEnv(),
resource.WithProcess(),
resource.WithTelemetrySDK(),
resource.WithHost(),
resource.WithAttributes(
semconv.ServiceNameKey.String(conf.ServiceName), // 应用名
semconv.ServiceVersionKey.String(conf.Version), // 应用版本
semconv.DeploymentEnvironmentKey.String(conf.Env), // 部署环境
semconv.HostNameKey.String(hostName), // 主机名
semconv.ServiceNameKey.String(o.config.ServiceName), // 应用名
semconv.ServiceVersionKey.String(o.config.Version), // 应用版本
semconv.DeploymentEnvironmentKey.String(o.config.Env), // 部署环境
semconv.HostNameKey.String(hostName), // 主机名
),
)
if err != nil {
return nil, err
}
return r, nil
return
}
func initHTTPExporterAndSpanProcessor(ctx context.Context, conf *Config) (*otlptrace.Exporter, sdktrace.SpanProcessor, error) {
opts := []otlptracehttp.Option{
otlptracehttp.WithInsecure(),
otlptracehttp.WithCompression(1),
func (o *OTEL) initMeterProvider(ctx context.Context) (err error) {
exporterGrpcFunc := func(ctx context.Context) (sdkmetric.Exporter, error) {
opts := []otlpmetricgrpc.Option{
otlpmetricgrpc.WithEndpoint(o.config.EndpointGRPC),
otlpmetricgrpc.WithCompressor(gzip.Name),
}
if o.config.Token != "" {
headers := map[string]string{"Authentication": o.config.Token}
opts = append(opts, otlpmetricgrpc.WithHeaders(headers))
}
exporter, err := otlpmetricgrpc.New(ctx, opts...)
if err != nil {
return nil, err
}
return exporter, nil
}
if conf.Token != "" {
opts = append(opts, otlptracehttp.WithURLPath(conf.Token))
exporterHttpFunc := func(ctx context.Context) (sdkmetric.Exporter, error) {
opts := []otlpmetrichttp.Option{
otlpmetrichttp.WithEndpoint(o.config.EndpointHTTP),
otlpmetrichttp.WithCompression(1),
}
if o.config.Token != "" {
opts = append(opts, otlpmetrichttp.WithURLPath(o.config.Token))
}
exporter, err := otlpmetrichttp.New(ctx, opts...)
if err != nil {
return nil, err
}
return exporter, nil
}
if conf.EndpointHTTP != "" {
opts = append(opts, otlptracehttp.WithEndpoint(conf.EndpointHTTP))
var exporter sdkmetric.Exporter
if o.config.EndpointHTTP != "" {
exporter, err = exporterHttpFunc(ctx)
} else {
exporter, err = exporterGrpcFunc(ctx)
}
traceExporter, err := otlptrace.New(ctx, otlptracehttp.NewClient(opts...))
if err != nil {
return nil, nil, err
return
}
batchSpanProcessor := sdktrace.NewBatchSpanProcessor(traceExporter)
meterProvider := sdkmetric.NewMeterProvider(
sdkmetric.WithReader(
sdkmetric.NewPeriodicReader(exporter),
),
sdkmetric.WithResource(o.Resource),
)
otel.SetMeterProvider(meterProvider)
return traceExporter, batchSpanProcessor, nil
container.AddCloseAble(func() {
if err := meterProvider.Shutdown(ctx); err != nil {
otel.Handle(err)
}
})
return
}
func initGrpcExporterAndSpanProcessor(ctx context.Context, conf *Config) (*otlptrace.Exporter, sdktrace.SpanProcessor, error) {
opts := []otlptracegrpc.Option{
otlptracegrpc.WithCompressor(gzip.Name),
func (o *OTEL) initTracerProvider(ctx context.Context) error {
exporterGrpcFunc := func(ctx context.Context) (*otlptrace.Exporter, error) {
opts := []otlptracegrpc.Option{
otlptracegrpc.WithCompressor(gzip.Name),
otlptracegrpc.WithEndpoint(o.config.EndpointGRPC),
}
if o.config.Token != "" {
headers := map[string]string{"Authentication": o.config.Token}
opts = append(opts, otlptracegrpc.WithHeaders(headers))
}
exporter, err := otlptrace.New(ctx, otlptracegrpc.NewClient(opts...))
if err != nil {
return nil, err
}
container.AddCloseAble(func() {
cxt, cancel := context.WithTimeout(ctx, time.Second)
defer cancel()
if err := exporter.Shutdown(cxt); err != nil {
otel.Handle(err)
}
})
return exporter, nil
}
if conf.Token != "" {
headers := map[string]string{"Authentication": conf.Token}
opts = append(opts, otlptracegrpc.WithHeaders(headers))
exporterHttpFunc := func(ctx context.Context) (*otlptrace.Exporter, error) {
opts := []otlptracehttp.Option{
otlptracehttp.WithInsecure(),
otlptracehttp.WithCompression(1),
otlptracehttp.WithEndpoint(o.config.EndpointHTTP),
}
if o.config.Token != "" {
opts = append(opts, otlptracehttp.WithURLPath(o.config.Token))
}
exporter, err := otlptrace.New(ctx, otlptracehttp.NewClient(opts...))
if err != nil {
return nil, err
}
return exporter, nil
}
if conf.EndpointGRPC != "" {
opts = append(opts, otlptracegrpc.WithEndpoint(conf.EndpointGRPC))
var exporter *otlptrace.Exporter
var err error
if o.config.EndpointHTTP != "" {
exporter, err = exporterHttpFunc(ctx)
} else {
exporter, err = exporterGrpcFunc(ctx)
}
traceExporter, err := otlptrace.New(ctx, otlptracegrpc.NewClient(opts...))
if err != nil {
return nil, nil, err
return err
}
batchSpanProcessor := sdktrace.NewBatchSpanProcessor(traceExporter)
traceProvider := sdktrace.NewTracerProvider(
sdktrace.WithSampler(sdktrace.AlwaysSample()),
sdktrace.WithBatcher(exporter),
sdktrace.WithResource(o.Resource),
sdktrace.WithSpanProcessor(sdktrace.NewBatchSpanProcessor(exporter)),
)
container.AddCloseAble(func() {
if err := traceProvider.Shutdown(ctx); err != nil {
otel.Handle(err)
}
})
return traceExporter, batchSpanProcessor, nil
otel.SetTracerProvider(traceProvider)
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
propagation.TraceContext{},
propagation.Baggage{},
))
return err
}