Skip to content

Commit 0d3b850

Browse files
pjanottijmsnll
authored andcommitted
[pkg/stanza] Cache event publishers: log warn once per provider (open-telemetry#27658)
**Description:** Cache the publisher event to: 1. Avoid logging the same error message every time one event from the given source is logged. 2. Avoid opening and closing the event publisher for every single event. **Link to tracking Issue:** [Item 4 described on the investigation](open-telemetry#21491 (comment)) for issue open-telemetry#21491. **Testing:** * Go tests for `pkg/stanza` and `receiver/windowseventlogreceiver` on Windows box. * Ran the contrib build locally to validate the change. * Can't run the full make locally: misspell is failing on Windows because the command line is too long. **Documentation:** Let me know if changing the severity of the log message requires a changelog update.
1 parent 1657d50 commit 0d3b850

File tree

6 files changed

+182
-7
lines changed

6 files changed

+182
-7
lines changed
Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
# Use this changelog template to create an entry for release notes.
2+
3+
# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
4+
change_type: enhancement
5+
6+
# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
7+
component: pkg/stanza
8+
9+
# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
10+
note: Log warning, instead of error, when Windows Event Log publisher metadata is not available and cache the successfully retrieved ones.
11+
12+
# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
13+
issues: [27658]
14+
15+
# (Optional) One or more lines of additional information to render under the primary note.
16+
# These lines will be padded with 2 spaces and then inserted directly into the document.
17+
# Use pipe (|) for multiline entries.
18+
subtext:
19+
20+
# If your change doesn't affect end users or the exported elements of any package,
21+
# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
22+
# Optional: The change log or logs in which this entry should be included.
23+
# e.g. '[user]' or '[user, api]'
24+
# Include 'user' if the change is relevant to end users.
25+
# Include 'api' if there is a change to a library API.
26+
# Default: '[user]'
27+
change_logs: []

pkg/stanza/operator/input/windows/operator.go

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@ type Input struct {
9494
excludeProviders []string
9595
pollInterval time.Duration
9696
persister operator.Persister
97+
publisherCache publisherCache
9798
cancel context.CancelFunc
9899
wg sync.WaitGroup
99100
}
@@ -123,6 +124,8 @@ func (e *Input) Start(persister operator.Persister) error {
123124
return fmt.Errorf("failed to open subscription: %w", err)
124125
}
125126

127+
e.publisherCache = newPublisherCache()
128+
126129
e.wg.Add(1)
127130
go e.readOnInterval(ctx)
128131
return nil
@@ -141,6 +144,10 @@ func (e *Input) Stop() error {
141144
return fmt.Errorf("failed to close bookmark: %w", err)
142145
}
143146

147+
if err := e.publisherCache.evictAll(); err != nil {
148+
return fmt.Errorf("failed to close publishers: %w", err)
149+
}
150+
144151
return nil
145152
}
146153

@@ -231,13 +238,15 @@ func (e *Input) processEvent(ctx context.Context, event Event) {
231238
}
232239
}
233240

234-
publisher := NewPublisher()
235-
if err := publisher.Open(simpleEvent.Provider.Name); err != nil {
236-
e.Errorf("Failed to open publisher: %s: writing log entry to pipeline without metadata", err)
241+
publisher, openPublisherErr := e.publisherCache.get(simpleEvent.Provider.Name)
242+
if openPublisherErr != nil {
243+
e.Warnf("Failed to open the %q event source, respective log entries can't be formatted: %s", simpleEvent.Provider.Name, openPublisherErr)
244+
}
245+
246+
if !publisher.Valid() {
237247
e.sendEvent(ctx, simpleEvent)
238248
return
239249
}
240-
defer publisher.Close()
241250

242251
formattedEvent, err := event.RenderFormatted(e.buffer, publisher)
243252
if err != nil {

pkg/stanza/operator/input/windows/publisher.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,10 @@ func (p *Publisher) Open(provider string) error {
3636
return nil
3737
}
3838

39+
func (p *Publisher) Valid() bool {
40+
return p.handle != 0
41+
}
42+
3943
// Close will close the publisher handle.
4044
func (p *Publisher) Close() error {
4145
if p.handle == 0 {

pkg/stanza/operator/input/windows/publisher_test.go

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ func TestPublisherOpenPreexisting(t *testing.T) {
1717
err := publisher.Open("")
1818
require.Error(t, err)
1919
require.Contains(t, err.Error(), "publisher handle is already open")
20+
require.True(t, publisher.Valid())
2021
}
2122

2223
func TestPublisherOpenInvalidUTF8(t *testing.T) {
@@ -25,44 +26,60 @@ func TestPublisherOpenInvalidUTF8(t *testing.T) {
2526
err := publisher.Open(invalidUTF8)
2627
require.Error(t, err)
2728
require.Contains(t, err.Error(), "failed to convert the provider name \"\\x00\" to utf16: invalid argument")
29+
require.False(t, publisher.Valid())
2830
}
2931

3032
func TestPublisherOpenSyscallFailure(t *testing.T) {
3133
publisher := NewPublisher()
3234
provider := "provider"
33-
openPublisherMetadataProc = SimpleMockProc(0, 0, ErrorNotSupported)
35+
defer mockWithDeferredRestore(&openPublisherMetadataProc, SimpleMockProc(0, 0, ErrorNotSupported))()
3436
err := publisher.Open(provider)
3537
require.Error(t, err)
3638
require.Contains(t, err.Error(), "failed to open the metadata for the \"provider\" provider: The request is not supported.")
39+
require.False(t, publisher.Valid())
3740
}
3841

3942
func TestPublisherOpenSuccess(t *testing.T) {
4043
publisher := NewPublisher()
4144
provider := "provider"
42-
openPublisherMetadataProc = SimpleMockProc(5, 0, ErrorSuccess)
45+
defer mockWithDeferredRestore(&openPublisherMetadataProc, SimpleMockProc(5, 0, ErrorSuccess))()
4346
err := publisher.Open(provider)
4447
require.NoError(t, err)
4548
require.Equal(t, uintptr(5), publisher.handle)
49+
require.True(t, publisher.Valid())
4650
}
4751

4852
func TestPublisherCloseWhenAlreadyClosed(t *testing.T) {
4953
publisher := NewPublisher()
5054
err := publisher.Close()
5155
require.NoError(t, err)
56+
require.False(t, publisher.Valid())
5257
}
5358

5459
func TestPublisherCloseSyscallFailure(t *testing.T) {
5560
publisher := Publisher{handle: 5}
56-
closeProc = SimpleMockProc(0, 0, ErrorNotSupported)
61+
defer mockWithDeferredRestore(&closeProc, SimpleMockProc(0, 0, ErrorNotSupported))()
5762
err := publisher.Close()
5863
require.Error(t, err)
5964
require.Contains(t, err.Error(), "failed to close publisher")
65+
require.True(t, publisher.Valid())
6066
}
6167

6268
func TestPublisherCloseSuccess(t *testing.T) {
6369
publisher := Publisher{handle: 5}
70+
originalCloseProc := closeProc
6471
closeProc = SimpleMockProc(1, 0, ErrorSuccess)
72+
defer func() { closeProc = originalCloseProc }()
6573
err := publisher.Close()
6674
require.NoError(t, err)
6775
require.Equal(t, uintptr(0), publisher.handle)
76+
require.False(t, publisher.Valid())
77+
}
78+
79+
func mockWithDeferredRestore(call *SyscallProc, mockCall SyscallProc) func() {
80+
original := *call
81+
*call = mockCall
82+
return func() {
83+
*call = original
84+
}
6885
}
Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
//go:build windows
5+
// +build windows
6+
7+
package windows // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/input/windows"
8+
9+
import (
10+
"errors"
11+
)
12+
13+
type publisherCache struct {
14+
cache map[string]Publisher
15+
}
16+
17+
func newPublisherCache() publisherCache {
18+
return publisherCache{
19+
cache: make(map[string]Publisher),
20+
}
21+
}
22+
23+
func (c *publisherCache) get(provider string) (publisher Publisher, openPublisherErr error) {
24+
publisher, ok := c.cache[provider]
25+
if ok {
26+
return publisher, nil
27+
}
28+
29+
publisher = NewPublisher()
30+
err := publisher.Open(provider)
31+
32+
// Always store the publisher even if there was an error opening it.
33+
c.cache[provider] = publisher
34+
35+
return publisher, err
36+
}
37+
38+
func (c *publisherCache) evictAll() error {
39+
var errs error
40+
for _, publisher := range c.cache {
41+
if publisher.Valid() {
42+
if err := publisher.Close(); err != nil {
43+
errs = errors.Join(errs, err)
44+
}
45+
}
46+
}
47+
48+
c.cache = make(map[string]Publisher)
49+
return errs
50+
}
Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
//go:build windows
5+
// +build windows
6+
7+
package windows
8+
9+
import (
10+
"testing"
11+
12+
"github.com/stretchr/testify/require"
13+
)
14+
15+
func TestGetValidPublisher(t *testing.T) {
16+
publisherCache := newPublisherCache()
17+
defer publisherCache.evictAll()
18+
19+
// Provider "Application" exists in all Windows versions.
20+
publisher, openPublisherErr := publisherCache.get("Application")
21+
require.NoError(t, openPublisherErr)
22+
require.True(t, publisher.Valid())
23+
24+
// Get the same publisher again.
25+
publisher, openPublisherErr = publisherCache.get("Application")
26+
require.NoError(t, openPublisherErr)
27+
require.True(t, publisher.Valid())
28+
}
29+
30+
func TestGetInvalidPublisher(t *testing.T) {
31+
publisherCache := newPublisherCache()
32+
defer publisherCache.evictAll()
33+
34+
// Provider "InvalidProvider" does not exist in any Windows version.
35+
publisher, openPublisherErr := publisherCache.get("InvalidProvider")
36+
require.Error(t, openPublisherErr, "%v", publisherCache)
37+
require.False(t, publisher.Valid())
38+
39+
// Get "InvalidProvider" publisher again.
40+
publisher, openPublisherErr = publisherCache.get("InvalidProvider")
41+
require.NoError(t, openPublisherErr) // It is cached, no error opening it.
42+
require.False(t, publisher.Valid())
43+
}
44+
45+
func TestValidAndInvalidPublishers(t *testing.T) {
46+
publisherCache := newPublisherCache()
47+
defer publisherCache.evictAll()
48+
49+
// Provider "Application" exists in all Windows versions.
50+
publisher, openPublisherErr := publisherCache.get("Application")
51+
require.NoError(t, openPublisherErr)
52+
require.True(t, publisher.Valid())
53+
54+
// Provider "InvalidProvider" does not exist in any Windows version.
55+
publisher, openPublisherErr = publisherCache.get("InvalidProvider")
56+
require.Error(t, openPublisherErr, "%v", publisherCache)
57+
require.False(t, publisher.Valid())
58+
59+
// Get the existing publisher again.
60+
publisher, openPublisherErr = publisherCache.get("Application")
61+
require.NoError(t, openPublisherErr)
62+
require.True(t, publisher.Valid())
63+
64+
// Get "InvalidProvider" publisher again.
65+
publisher, openPublisherErr = publisherCache.get("InvalidProvider")
66+
require.NoError(t, openPublisherErr) // It is cached, no error opening it.
67+
require.False(t, publisher.Valid())
68+
}

0 commit comments

Comments
 (0)