Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[jaeger-v2] Add StartTimeMin and StartTimeMax params to consumeTraces function. #5288

Merged
merged 4 commits into from
Mar 23, 2024
Merged

[jaeger-v2] Add StartTimeMin and StartTimeMax params to consumeTraces function. #5288

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4cfcb12
fix JaegerV2 storage receiver consumeTraces query params
james-ryans Mar 22, 2024
3fca599
use config file telemetry log instead of creating one
james-ryans Mar 22, 2024
de57e56
remove unused logger that passed to newDataReceiver
james-ryans Mar 22, 2024
755dc7e
pass the whole TelemetrySettings instead of only a Logger
james-ryans Mar 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 24 additions & 14 deletions cmd/jaeger/internal/integration/datareceivers/jaegerstorage.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,50 +6,60 @@ package datareceivers
import (
"context"
"fmt"
"time"

"github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage/storagetest"
"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/testbed"
"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/component/componenttest"
"go.opentelemetry.io/collector/consumer"
"go.opentelemetry.io/collector/extension"
"go.opentelemetry.io/collector/receiver"
"go.opentelemetry.io/collector/receiver/receivertest"

"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/jaegerstorage"
"github.com/jaegertracing/jaeger/cmd/jaeger/internal/integration/receivers/storagereceiver"
)

type jaegerStorageDataReceiver struct {
TraceStorage string
StorageConfig *jaegerstorage.Config
host *storagetest.StorageHost
receiver receiver.Traces
TelemetrySettings component.TelemetrySettings
TraceStorage string
StorageConfig *jaegerstorage.Config
host *storagetest.StorageHost
receiver receiver.Traces
}

func NewJaegerStorageDataReceiver(traceStorage string, storageConfig *jaegerstorage.Config) testbed.DataReceiver {
func NewJaegerStorageDataReceiver(
telemetrySettings component.TelemetrySettings,
traceStorage string,
storageConfig *jaegerstorage.Config,
) testbed.DataReceiver {
return &jaegerStorageDataReceiver{
TraceStorage: traceStorage,
StorageConfig: storageConfig,
TelemetrySettings: telemetrySettings,
TraceStorage: traceStorage,
StorageConfig: storageConfig,
}
}

func (dr *jaegerStorageDataReceiver) Start(tc consumer.Traces, _ consumer.Metrics, _ consumer.Logs) error {
ctx := context.Background()

extSet := extension.CreateSettings{
ID: jaegerstorage.ID,
TelemetrySettings: dr.TelemetrySettings,
}
extFactory := jaegerstorage.NewFactory()
ext, err := extFactory.CreateExtension(ctx, extension.CreateSettings{
TelemetrySettings: componenttest.NewNopTelemetrySettings(),
BuildInfo: component.NewDefaultBuildInfo(),
}, dr.StorageConfig)
ext, err := extFactory.CreateExtension(ctx, extSet, dr.StorageConfig)
if err != nil {
return err
}

rcvSet := receivertest.NewNopCreateSettings()
rcvSet := receiver.CreateSettings{
ID: storagereceiver.ID,
TelemetrySettings: dr.TelemetrySettings,
}
rcvFactory := storagereceiver.NewFactory()
rcvCfg := rcvFactory.CreateDefaultConfig().(*storagereceiver.Config)
rcvCfg.TraceStorage = dr.TraceStorage
rcvCfg.PullInterval = 100 * time.Millisecond
rcv, err := rcvFactory.CreateTracesReceiver(ctx, rcvSet, rcvCfg, tc)
if err != nil {
return err
Expand Down
14 changes: 13 additions & 1 deletion cmd/jaeger/internal/integration/integration.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,11 @@ import (

"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/testbed"
"github.com/stretchr/testify/require"
"go.opentelemetry.io/collector/component/componenttest"
"go.opentelemetry.io/collector/confmap"
"go.opentelemetry.io/collector/confmap/provider/fileprovider"
"go.opentelemetry.io/collector/otelcol"
"go.opentelemetry.io/collector/service/telemetry"

"github.com/jaegertracing/jaeger/cmd/jaeger/internal"
"github.com/jaegertracing/jaeger/cmd/jaeger/internal/exporters/storageexporter"
Expand Down Expand Up @@ -46,13 +48,23 @@ func (s *StorageIntegration) newDataReceiver(t *testing.T, factories otelcol.Fac
cfg, err := configProvider.Get(context.Background(), factories)
require.NoError(t, err)

tel, err := telemetry.New(context.Background(), telemetry.Settings{}, cfg.Service.Telemetry)
require.NoError(t, err)

storageCfg, ok := cfg.Extensions[jaegerstorage.ID].(*jaegerstorage.Config)
require.True(t, ok, "no jaeger storage extension found in the config")

exporterCfg, ok := cfg.Exporters[storageexporter.ID].(*storageexporter.Config)
require.True(t, ok, "no jaeger storage exporter found in the config")

receiver := datareceivers.NewJaegerStorageDataReceiver(exporterCfg.TraceStorage, storageCfg)
telemetrySettings := componenttest.NewNopTelemetrySettings()
telemetrySettings.Logger = tel.Logger()

receiver := datareceivers.NewJaegerStorageDataReceiver(
telemetrySettings,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason why can't pass tel directly?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because tel (or telemetry.Telemetry) and component.TelemetrySettings is two different component.
telemetry.Telemetry initialize the zap.Logger from the config.
component.TelemetrySettings is the struct that contains zap.Logger that is used by components in OTEL to report telemetry matters.

This means that the extension and receiver in the data receiver are using component.TelemetrySettings to write logs. Based on your suggestion, I thought it is better if we pass the initialized struct and use it directly in the data receiver.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. Looking through OTEL code the situation seems to be quite messy, I don't know if they have a ticket to clean it up. E.g.

  • why are Telemetry and TelemetrySettings two different structs? They seem to server the same purpose (the latter being more comprehensive)
  • TelemetrySettings comment also mentions yet another similar struct servicetelemetry.TelemetrySettings

exporterCfg.TraceStorage,
storageCfg,
)
return receiver
}

Expand Down
3 changes: 3 additions & 0 deletions cmd/jaeger/internal/integration/receivers/storagereceiver/factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@ import (
// componentType is the name of this extension in configuration.
const componentType = component.Type("jaeger_storage_receiver")

// ID is the identifier of this extension.
var ID = component.NewID(componentType)

func NewFactory() receiver.Factory {
return receiver.NewFactory(
componentType,
Expand Down
5 changes: 4 additions & 1 deletion cmd/jaeger/internal/integration/receivers/storagereceiver/receiver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,8 +88,11 @@ func (r *storageReceiver) consumeLoop(ctx context.Context) error {
}

func (r *storageReceiver) consumeTraces(ctx context.Context, serviceName string) error {
endTime := time.Now()
traces, err := r.spanReader.FindTraces(ctx, &spanstore.TraceQueryParameters{
ServiceName: serviceName,
ServiceName: serviceName,
StartTimeMin: endTime.Add(-1 * time.Hour),
StartTimeMax: endTime,
})
if err != nil {
return err
Expand Down
13 changes: 5 additions & 8 deletions cmd/jaeger/internal/integration/receivers/storagereceiver/receiver_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import (
"github.com/jaegertracing/jaeger/model"
"github.com/jaegertracing/jaeger/storage"
factoryMocks "github.com/jaegertracing/jaeger/storage/mocks"
"github.com/jaegertracing/jaeger/storage/spanstore"
spanStoreMocks "github.com/jaegertracing/jaeger/storage/spanstore/mocks"
)

Expand Down Expand Up @@ -254,13 +253,11 @@ func TestReceiver_StartConsume(t *testing.T) {
t.Run(test.name, func(t *testing.T) {
reader := new(spanStoreMocks.Reader)
reader.On("GetServices", mock.AnythingOfType("*context.cancelCtx")).Return(test.services, nil)
for _, service := range test.services {
reader.On(
"FindTraces",
mock.AnythingOfType("*context.cancelCtx"),
&spanstore.TraceQueryParameters{ServiceName: service},
).Return(test.traces, test.tracesErr)
}
reader.On(
"FindTraces",
mock.AnythingOfType("*context.cancelCtx"),
mock.AnythingOfType("*spanstore.TraceQueryParameters"),
).Return(test.traces, test.tracesErr)
r.receiver.spanReader = reader

require.NoError(t, r.receiver.Shutdown(ctx))
Expand Down
Loading