eric/onix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feat: update the pr as per comment

Browse Source
This commit is contained in:
Manendra Pal Singh
2026-02-27 19:05:09 +05:30
parent 1293c241f7
commit 59aa058920
13 changed files with 159 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/log/log_test.go
View File

@@ -250,11 +250,11 @@ func TestError(t *testing.T) {
func TestRequest(t *testing.T) {
logPath := setupLogger(t, InfoLevel)
ctx := context.WithValue(context.Background(), requestID, "abc-123")
ctx = context.WithValue(context.Background(), transaction_id, "transaction-id-123-")
ctx = context.WithValue(context.Background(), message_id, "message-id-123")
ctx = context.WithValue(context.Background(), subscriber_id, "subscriber-id-123")
ctx = context.WithValue(context.Background(), module_id, "module-id-123")
ctx = context.WithValue(context.Background(), parent_id, "parent-id-123")
ctx = context.WithValue(ctx, transaction_id, "transaction-id-123-")
ctx = context.WithValue(ctx, message_id, "message-id-123")
ctx = context.WithValue(ctx, subscriber_id, "subscriber-id-123")
ctx = context.WithValue(ctx, module_id, "module-id-123")
ctx = context.WithValue(ctx, parent_id, "parent-id-123")
req, _ := http.NewRequest("POST", "/api/test", bytes.NewBuffer([]byte(`{"key":"value"}`)))
req.RemoteAddr = "127.0.0.1:8080"

4
pkg/model/model.go
View File

@@ -58,7 +58,7 @@ const (
ContextKeyParentID ContextKey = "parent_id"
// ContextKeyRemoteID is the context key for the caller who is calling the bap/bpp
ContextKeyRemoteID ContextKey = "caller_id"
ContextKeyRemoteID ContextKey = "remote_id"
)
var contextKeys = map[string]ContextKey{
@@ -67,7 +67,7 @@ var contextKeys = map[string]ContextKey{
"subscriber_id": ContextKeySubscriberID,
"module_id": ContextKeyModuleID,
"parent_id": ContextKeyParentID,
"caller_id": ContextKeyRemoteID,
"remote_id": ContextKeyRemoteID,
}
// ParseContextKey converts a string into a valid ContextKey.

22
pkg/plugin/implementation/cache/cache.go vendored
View File

@@ -114,22 +114,25 @@ func New(ctx context.Context, cfg *Config) (*Cache, func() error, error) {
// Get retrieves the value for the specified key from Redis.
func (c *Cache) Get(ctx context.Context, key string) (string, error) {
result, err := c.Client.Get(ctx, key).Result()
tracer := otel.Tracer(telemetry.ScopeName, trace.WithInstrumentationVersion(telemetry.ScopeVersion))
spanCtx, span := tracer.Start(ctx, "redis_get")
defer span.End()
result, err := c.Client.Get(spanCtx, key).Result()
if c.metrics != nil {
attrs := []attribute.KeyValue{
telemetry.AttrOperation.String("get"),
}
switch {
case err == redis.Nil:
c.metrics.CacheMissesTotal.Add(ctx, 1, metric.WithAttributes(attrs...))
c.metrics.CacheOperationsTotal.Add(ctx, 1,
c.metrics.CacheMissesTotal.Add(spanCtx, 1, metric.WithAttributes(attrs...))
c.metrics.CacheOperationsTotal.Add(spanCtx, 1,
metric.WithAttributes(append(attrs, telemetry.AttrStatus.String("miss"))...))
case err != nil:
c.metrics.CacheOperationsTotal.Add(ctx, 1,
c.metrics.CacheOperationsTotal.Add(spanCtx, 1,
metric.WithAttributes(append(attrs, telemetry.AttrStatus.String("error"))...))
default:
c.metrics.CacheHitsTotal.Add(ctx, 1, metric.WithAttributes(attrs...))
c.metrics.CacheOperationsTotal.Add(ctx, 1,
c.metrics.CacheHitsTotal.Add(spanCtx, 1, metric.WithAttributes(attrs...))
c.metrics.CacheOperationsTotal.Add(spanCtx, 1,
metric.WithAttributes(append(attrs, telemetry.AttrStatus.String("hit"))...))
}
}
@@ -149,8 +152,11 @@ func (c *Cache) Set(ctx context.Context, key, value string, ttl time.Duration) e
// Delete removes the specified key from Redis.
func (c *Cache) Delete(ctx context.Context, key string) error {
err := c.Client.Del(ctx, key).Err()
c.recordOperation(ctx, "delete", err)
tracer := otel.Tracer(telemetry.ScopeName, trace.WithInstrumentationVersion(telemetry.ScopeVersion))
spanCtx, span := tracer.Start(ctx, "redis_delete")
defer span.End()
err := c.Client.Del(spanCtx, key).Err()
c.recordOperation(spanCtx, "delete", err)
return err
}

86
pkg/plugin/implementation/otelsetup/otelsetup.go
View File

@@ -30,18 +30,19 @@ type Setup struct{}
// Config represents OpenTelemetry related configuration.
type Config struct {
ServiceName string `yaml:"serviceName"`
ServiceVersion string `yaml:"serviceVersion"`
Environment string `yaml:"environment"`
Domain string `yaml:"domain"`
DeviceID string `yaml:"deviceID"`
EnableMetrics bool `yaml:"enableMetrics"`
EnableTracing bool `yaml:"enableTracing"`
EnableLogs bool `yaml:"enableLogs"`
OtlpEndpoint string `yaml:"otlpEndpoint"`
TimeInterval int64 `yaml:"timeInterval"`
Producer string `yaml:"producer"`
ProducerType string `yaml:"producerType"`
ServiceName string `yaml:"serviceName"`
ServiceVersion string `yaml:"serviceVersion"`
Environment string `yaml:"environment"`
Domain string `yaml:"domain"`
DeviceID string `yaml:"deviceID"`
EnableMetrics bool `yaml:"enableMetrics"`
EnableTracing bool `yaml:"enableTracing"`
EnableLogs bool `yaml:"enableLogs"`
OtlpEndpoint string `yaml:"otlpEndpoint"`
TimeInterval int64 `yaml:"timeInterval"`
AuditFieldsConfig string `yaml:"auditFieldsConfig"`
Producer string `yaml:"producer"`
ProducerType string `yaml:"producerType"`
}
// DefaultConfig returns sensible defaults for telemetry configuration.
@@ -62,13 +63,17 @@ func ToPluginConfig(cfg *Config) *plugin.Config {
return &plugin.Config{
ID: "otelsetup",
Config: map[string]string{
"serviceName": cfg.ServiceName,
"serviceVersion": cfg.ServiceVersion,
"environment": cfg.Environment,
"enableMetrics": fmt.Sprintf("%t", cfg.EnableMetrics),
"enableTracing": fmt.Sprintf("%t", cfg.EnableTracing),
"otelEndpoint": cfg.OtlpEndpoint,
"deviceID": cfg.DeviceID,
"serviceName": cfg.ServiceName,
"serviceVersion": cfg.ServiceVersion,
"environment": cfg.Environment,
"domain": cfg.Domain,
"enableMetrics": fmt.Sprintf("%t", cfg.EnableMetrics),
"enableTracing": fmt.Sprintf("%t", cfg.EnableTracing),
"enableLogs": fmt.Sprintf("%t", cfg.EnableLogs),
"otlpEndpoint": cfg.OtlpEndpoint,
"deviceID": cfg.DeviceID,
"timeInterval": fmt.Sprintf("%d", cfg.TimeInterval),
"auditFieldsConfig": cfg.AuditFieldsConfig,
},
}
}
@@ -101,16 +106,13 @@ func (Setup) New(ctx context.Context, cfg *Config) (*telemetry.Provider, error)
cfg.TimeInterval = DefaultConfig().TimeInterval
}
if !cfg.EnableMetrics && !cfg.EnableTracing {
log.Info(ctx, "OpenTelemetry metrics and tracing are disabled")
if !cfg.EnableMetrics && !cfg.EnableTracing && !cfg.EnableLogs {
log.Info(ctx, "OpenTelemetry metrics, tracing, and logs are all disabled")
return &telemetry.Provider{
Shutdown: func(context.Context) error { return nil },
}, nil
}
//this will be used by both metric and traces
// to build resource with envelope metadata
baseAttrs := []attribute.KeyValue{
attribute.String("service.name", cfg.ServiceName),
attribute.String("service.version", cfg.ServiceVersion),
@@ -121,53 +123,49 @@ func (Setup) New(ctx context.Context, cfg *Config) (*telemetry.Provider, error)
attribute.String("producer", cfg.Producer),
}
resMetric, err := resource.New(ctx, resource.WithAttributes(buildAtts(baseAttrs, "METRIC")...))
if err != nil {
return nil, fmt.Errorf("failed to create telemetry resource for metric: %w", err)
}
//OTLP metric
var meterProvider *metric.MeterProvider
if cfg.EnableMetrics {
metricExpoter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithEndpoint(cfg.OtlpEndpoint),
resMetric, err := resource.New(ctx, resource.WithAttributes(buildAtts(baseAttrs, "METRIC")...))
if err != nil {
return nil, fmt.Errorf("failed to create telemetry resource for metric: %w", err)
}
metricExporter, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithEndpoint(cfg.OtlpEndpoint),
otlpmetricgrpc.WithDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())))
if err != nil {
return nil, fmt.Errorf("failed to create OTLP metric exporter: %w", err)
}
reader := metric.NewPeriodicReader(metricExpoter, metric.WithInterval(time.Second*time.Duration(cfg.TimeInterval)))
reader := metric.NewPeriodicReader(metricExporter, metric.WithInterval(time.Second*time.Duration(cfg.TimeInterval)))
meterProvider = metric.NewMeterProvider(metric.WithReader(reader), metric.WithResource(resMetric))
otel.SetMeterProvider(meterProvider)
log.Infof(ctx, "OpenTelemetry metrics initialized for service=%s version=%s env=%s (OTLP endpoint=%s)",
cfg.ServiceName, cfg.ServiceVersion, cfg.Environment, cfg.OtlpEndpoint)
// for the go runtime metrics
if err := runtime.Start(runtime.WithMinimumReadMemStatsInterval(runtime.DefaultMinimumReadMemStatsInterval)); err != nil {
log.Warnf(ctx, "Failed to start Go runtime instrumentation: %v", err)
}
}
//OTLP traces
restrace, err := resource.New(ctx, resource.WithAttributes(buildAtts(baseAttrs, "API")...))
if err != nil {
return nil, fmt.Errorf("failed to create trace resource: %w", err)
}
var traceProvider *trace.TracerProvider
if cfg.EnableTracing {
traceExpoter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithEndpoint(cfg.OtlpEndpoint), otlptracegrpc.WithDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())))
resTrace, err := resource.New(ctx, resource.WithAttributes(buildAtts(baseAttrs, "API")...))
if err != nil {
return nil, fmt.Errorf("failed to create trace resource: %w", err)
}
traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithEndpoint(cfg.OtlpEndpoint), otlptracegrpc.WithDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())))
if err != nil {
return nil, fmt.Errorf("failed to create OTLP trace exporter: %w", err)
}
traceProvider = trace.NewTracerProvider(trace.WithBatcher(traceExpoter), trace.WithResource(restrace)) //TODO: need to add the trace sampleing rate
traceProvider = trace.NewTracerProvider(trace.WithBatcher(traceExporter), trace.WithResource(resTrace))
otel.SetTracerProvider(traceProvider)
log.Infof(ctx, "OpenTelemetry tracing initialized for service=%s (OTLP endpoint=%s)",
cfg.ServiceName, cfg.OtlpEndpoint)
}
resAudit, err := resource.New(ctx, resource.WithAttributes(buildAtts(baseAttrs, "AUDIT")...))
if err != nil {
return nil, fmt.Errorf("failed to create audit resource: %w", err)
}
var logProvider *logsdk.LoggerProvider
if cfg.EnableLogs {
resAudit, err := resource.New(ctx, resource.WithAttributes(buildAtts(baseAttrs, "AUDIT")...))
if err != nil {
return nil, fmt.Errorf("failed to create audit resource: %w", err)
}
logExporter, err := otlploggrpc.New(ctx, otlploggrpc.WithEndpoint(cfg.OtlpEndpoint), otlploggrpc.WithDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())))
if err != nil {
return nil, fmt.Errorf("failed to create OTLP logs exporter: %w", err)

57
pkg/plugin/implementation/otelsetup/otelsetup_test.go
View File

@@ -169,6 +169,7 @@ func TestToPluginConfig_Success(t *testing.T) {
ServiceVersion: "1.0.0",
EnableMetrics: true,
EnableTracing: true,
EnableLogs: true,
Environment: "test",
Domain: "test-domain",
DeviceID: "test-device",
@@ -177,13 +178,17 @@ func TestToPluginConfig_Success(t *testing.T) {
},
expectedID: "otelsetup",
expectedConfig: map[string]string{
"serviceName": "test-service",
"serviceVersion": "1.0.0",
"environment": "test",
"enableMetrics": "true",
"enableTracing": "true",
"otelEndpoint": "localhost:4317",
"deviceID": "test-device",
"serviceName": "test-service",
"serviceVersion": "1.0.0",
"environment": "test",
"domain": "test-domain",
"enableMetrics": "true",
"enableTracing": "true",
"enableLogs": "true",
"otlpEndpoint": "localhost:4317",
"deviceID": "test-device",
"timeInterval": "5",
"auditFieldsConfig": "",
},
},
{
@@ -197,13 +202,17 @@ func TestToPluginConfig_Success(t *testing.T) {
},
expectedID: "otelsetup",
expectedConfig: map[string]string{
"serviceName": "my-service",
"serviceVersion": "2.0.0",
"environment": "production",
"enableMetrics": "false",
"enableTracing": "false",
"otelEndpoint": "",
"deviceID": "",
"serviceName": "my-service",
"serviceVersion": "2.0.0",
"environment": "production",
"domain": "",
"enableMetrics": "false",
"enableTracing": "false",
"enableLogs": "false",
"otlpEndpoint": "",
"deviceID": "",
"timeInterval": "0",
"auditFieldsConfig": "",
},
},
{
@@ -220,13 +229,17 @@ func TestToPluginConfig_Success(t *testing.T) {
},
expectedID: "otelsetup",
expectedConfig: map[string]string{
"serviceName": "",
"serviceVersion": "",
"environment": "",
"enableMetrics": "true",
"enableTracing": "false",
"otelEndpoint": "",
"deviceID": "",
"serviceName": "",
"serviceVersion": "",
"environment": "",
"domain": "",
"enableMetrics": "true",
"enableTracing": "false",
"enableLogs": "false",
"otlpEndpoint": "",
"deviceID": "",
"timeInterval": "0",
"auditFieldsConfig": "",
},
},
}
@@ -298,7 +311,7 @@ func TestToPluginConfig_BooleanConversion(t *testing.T) {
require.NotNil(t, result)
assert.Equal(t, tt.expectedMetric, result.Config["enableMetrics"], "enableMetrics should be converted to string correctly")
assert.Equal(t, tt.expectedTrace, result.Config["enableTracing"], "enableTracing should be converted to string correctly")
assert.Equal(t, "localhost:4317", result.Config["otelEndpoint"], "otelEndpoint should be included")
assert.Equal(t, "localhost:4317", result.Config["otlpEndpoint"], "otlpEndpoint should be included")
assert.Equal(t, "test-device", result.Config["deviceID"], "deviceID should be included")
})
}

2
pkg/telemetry/audit.go
View File

@@ -23,8 +23,6 @@ func EmitAuditLogs(ctx context.Context, body []byte, attrs ...log.KeyValue) {
return
}
//maskedBody := MaskPIIInAuditBody(body)
sum := sha256.Sum256(body)
auditBody := selectAuditPayload(ctx, body)
auditlog := provider.Logger(auditLoggerName)

4
pkg/telemetry/pluginMetrics.go
View File

@@ -57,8 +57,8 @@ var (
var (
networkMetricsCfgMu sync.RWMutex
networkMetricsGranularity = "10mim" // default
networkMetricsFrequency = "10mim" // default
networkMetricsGranularity = "10min" // default
networkMetricsFrequency = "10min" // default
)
func SetNetworkMetricsConfig(granularity, frequency string) {