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

Jump to bottom

Discovery observers Start failures should not stop nor crash the collector #5299

Merged
merged 4 commits into from
Aug 30, 2024
Merged

Discovery observers Start failures should not stop nor crash the collector #5299

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
036f6fb
Discovery observers Start failures should not stop nor crash the coll…
pjanotti Aug 30, 2024
f797865
Changelog
pjanotti Aug 30, 2024
356b2c3
fieldaligment for discoverer_test.go
pjanotti Aug 30, 2024
07a5bff
Add comment to GetExtensions
pjanotti Aug 30, 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
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,10 @@

- (Splunk) Deprecate the nagios monitor ([#5172](https://github.com/signalfx/splunk-otel-collector/pull/5172))

### 🧰 Bug fixes 🧰

- (Splunk) Discovery observers start failures should not stop the collector ([#5299](https://github.com/signalfx/splunk-otel-collector/pull/5299))

## v0.108.0

This Splunk OpenTelemetry Collector release includes changes from the [opentelemetry-collector v0.108.1](https://github.com/open-telemetry/opentelemetry-collector/releases/tag/v0.108.1) and the [opentelemetry-collector-contrib v0.108.0](https://github.com/open-telemetry/opentelemetry-collector-contrib/releases/tag/v0.108.0) releases where appropriate.
Expand Down
23 changes: 10 additions & 13 deletions internal/confmapprovider/discovery/discoverer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ type discoverer struct {
factories otelcol.Factories
// receiverID -> observerID -> config
unexpandedReceiverEntries map[component.ID]map[component.ID]map[string]any
extensions map[component.ID]otelcolextension.Extension
operationalObservers map[component.ID]otelcolextension.Extension // Only extensions successfully started should be added to this map.
logger *zap.Logger
discoveredReceivers map[component.ID]discovery.StatusType
configs map[string]*Config
Expand Down Expand Up @@ -104,7 +104,6 @@ func newDiscoverer(logger *zap.Logger) (*discoverer, error) {
logger: logger,
info: info,
factories: factories,
extensions: map[component.ID]otelcolextension.Extension{},
configs: map[string]*Config{},
duration: duration,
mu: sync.Mutex{},
Expand Down Expand Up @@ -182,10 +181,7 @@ func (d *discoverer) discover(cfg *Config) (map[string]any, error) {
return nil, nil
}

err = d.performDiscovery(discoveryReceivers, discoveryObservers)
if err != nil {
return nil, err
}
d.performDiscovery(discoveryReceivers, discoveryObservers)

discoveryConfig, err := d.discoveryConfig(cfg)
if err != nil {
Expand All @@ -194,7 +190,7 @@ func (d *discoverer) discover(cfg *Config) (map[string]any, error) {
return discoveryConfig, nil
}

func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelcolreceiver.Logs, discoveryObservers map[component.ID]otelcolextension.Extension) error {
func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelcolreceiver.Logs, discoveryObservers map[component.ID]otelcolextension.Extension) {
var cancels []context.CancelFunc

defer func() {
Expand All @@ -203,6 +199,8 @@ func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelco
}
}()

d.operationalObservers = make(map[component.ID]component.Component, len(discoveryObservers))

for observerID, observer := range discoveryObservers {
d.logger.Debug(fmt.Sprintf("starting observer %q", observerID))
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
Expand All @@ -212,7 +210,7 @@ func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelco
fmt.Sprintf("%q startup failed. Won't proceed with %q-based discovery", observerID, observerID.Type()),
zap.Error(e),
)
return e
continue
}
defer func(obsID component.ID, obsExt otelcolextension.Extension) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
Expand All @@ -221,6 +219,7 @@ func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelco
d.logger.Warn(fmt.Sprintf("error shutting down observer %q", obsID), zap.Error(e))
}
}(observerID, observer)
d.operationalObservers[observerID] = observer
}

for receiverID, receiver := range discoveryReceivers {
Expand All @@ -232,7 +231,7 @@ func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelco
fmt.Sprintf("%q startup failed.", receiverID),
zap.Error(err),
)
return err
continue
}
defer func(rcvID component.ID, rcv otelcolreceiver.Logs) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
Expand All @@ -249,8 +248,6 @@ func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelco
case <-context.Background().Done():
}
_, _ = fmt.Fprintf(os.Stderr, "Discovery complete.\n")

return nil
}

func (d *discoverer) createDiscoveryReceiversAndObservers(cfg *Config) (map[component.ID]otelcolreceiver.Logs, map[component.ID]otelcolextension.Extension, error) {
Expand All @@ -268,7 +265,6 @@ func (d *discoverer) createDiscoveryReceiversAndObservers(cfg *Config) (map[comp
// disabled by property
continue
}
d.extensions[observerID] = observer
discoveryObservers[observerID] = observer

discoveryReceiverDefaultConfig := discoveryReceiverFactory.CreateDefaultConfig()
Expand Down Expand Up @@ -627,8 +623,9 @@ func (d *discoverer) GetFactory(kind component.Kind, componentType component.Typ
}

// GetExtensions is a component.Host method used to forward discovery observers.
// This method only returns operational extensions, i.e., those that have been successfully started.
func (d *discoverer) GetExtensions() map[component.ID]otelcolextension.Extension {
return d.extensions
return d.operationalObservers
}

// GetExporters is a component.Host method.
Expand Down
22 changes: 7 additions & 15 deletions internal/confmapprovider/discovery/discoverer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,9 @@ import (

func TestInnerDiscoveryExecution(t *testing.T) {
tests := []struct {
name string
observers map[component.ID]otelcolextension.Extension
receivers map[component.ID]otelcolreceiver.Logs
expectedMsg string
observers map[component.ID]otelcolextension.Extension
receivers map[component.ID]otelcolreceiver.Logs
name string
}{
{
name: "happy_path",
Expand All @@ -62,7 +61,6 @@ func TestInnerDiscoveryExecution(t *testing.T) {
receivers: map[component.ID]otelcolreceiver.Logs{
component.MustNewID("receiver00"): &mockReceiverLogs{},
},
expectedMsg: "extension_start_error",
},
{
name: "fail_start_receiver",
Expand All @@ -75,10 +73,9 @@ func TestInnerDiscoveryExecution(t *testing.T) {
component.MustNewID("receiver01"): &mockReceiverLogs{mockComponent{startErr: fmt.Errorf("receiver_start_error")}},
component.MustNewID("receiver02"): &mockReceiverLogs{},
},
expectedMsg: "receiver_start_error",
},
{
name: "fail_shutdown_no_error_msg",
name: "fail_shutdown_no_crash",
observers: map[component.ID]otelcolextension.Extension{
component.MustNewID("observer00"): &mockExtension{},
component.MustNewID("observer01"): &mockExtension{},
Expand All @@ -98,14 +95,15 @@ func TestInnerDiscoveryExecution(t *testing.T) {
require.NotNil(t, d)

d.duration = 1 * time.Second
err = d.performDiscovery(tt.receivers, tt.observers)
d.performDiscovery(tt.receivers, tt.observers)

for _, observer := range tt.observers {
for id, observer := range tt.observers {
mockExtension := observer.(*mockExtension)
if mockExtension.started {
assert.True(t, mockExtension.shutdown)
} else {
assert.False(t, mockExtension.shutdown)
assert.NotContains(t, d.operationalObservers, id)
}
}

Expand All @@ -117,12 +115,6 @@ func TestInnerDiscoveryExecution(t *testing.T) {
assert.False(t, mockReceiver.shutdown)
}
}

if tt.expectedMsg != "" {
assert.ErrorContains(t, err, tt.expectedMsg)
} else {
assert.NoError(t, err)
}
})
}
}
Expand Down
Loading