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

Fix telemetry api subscription 403 forbidden error #1452

Merged
merged 2 commits into from
Aug 16, 2024
Merged

Fix telemetry api subscription 403 forbidden error #1452

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f7ae94e
Fixed telemetry api receiver subscription errors
jerrytfleung Jul 24, 2024
7c59bb5
Fixed tests
jerrytfleung Jul 24, 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
13 changes: 6 additions & 7 deletions collector/internal/lifecycle/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,12 +86,6 @@ func NewManager(ctx context.Context, logger *zap.Logger, version string) (contex
listener: listener,
}

go func() {
if err := lm.processEvents(ctx); err != nil {
lm.logger.Warn("Failed to process events", zap.Error(err))
}
}()

factories, _ := lambdacomponents.Components(res.ExtensionID)
lm.collector = collector.NewCollector(logger, factories, version)

Expand All @@ -107,12 +101,17 @@ func (lm *manager) Run(ctx context.Context) error {
return err
}

lm.wg.Add(1)
go func() {
if err := lm.processEvents(ctx); err != nil {
lm.logger.Warn("Failed to process events", zap.Error(err))
}
}()
lm.wg.Wait()
return nil
}

func (lm *manager) processEvents(ctx context.Context) error {
lm.wg.Add(1)
defer lm.wg.Done()

for {
Expand Down
19 changes: 17 additions & 2 deletions collector/internal/lifecycle/manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,18 @@ func (c *MockCollector) Stop() error {
func TestRun(t *testing.T) {
logger := zaptest.NewLogger(t)
ctx := context.Background()

server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
_, err := w.Write([]byte(`{"time":"2006-01-02T15:04:05.000Z", "eventType":"SHUTDOWN", "record":{}}`))
require.NoError(t, err)
_, err = io.ReadAll(r.Body)
require.NoError(t, err, "failed to read request body: %v", err)
}))
defer server.Close()
u, err := url.Parse(server.URL)
require.NoError(t, err)

// test with an error
lm := manager{
collector: &MockCollector{err: fmt.Errorf("test start error")},
Expand All @@ -55,14 +67,16 @@ func TestRun(t *testing.T) {
lm = manager{
collector: &MockCollector{},
logger: logger,
extensionClient: extensionapi.NewClient(logger, ""),
listener: telemetryapi.NewListener(logger),
extensionClient: extensionapi.NewClient(logger, u.Host),
}
require.NoError(t, lm.Run(ctx))
// test with waitgroup counter incremented
lm = manager{
collector: &MockCollector{},
logger: logger,
extensionClient: extensionapi.NewClient(logger, ""),
listener: telemetryapi.NewListener(logger),
extensionClient: extensionapi.NewClient(logger, u.Host),
}
lm.wg.Add(1)
go func() {
Expand Down Expand Up @@ -130,6 +144,7 @@ func TestProcessEvents(t *testing.T) {
listener: telemetryapi.NewListener(logger),
extensionClient: extensionapi.NewClient(logger, u.Host),
}
lm.wg.Add(1)
if tc.err != nil {
err = lm.processEvents(ctx)
require.Error(t, err)
Expand Down
Loading