feat: refactored for cloud logs service (#4973)

* feat: refactored for cloud logs service * fix: commment * fix: renamed mapper * fix: panic when trying to stop second time
kubeshop · Feb 6, 2024 · 0a286e2 · 0a286e2
1 parent 1bc0381
commit 0a286e2
Show file tree

Hide file tree

Showing 6 changed files with 347 additions and 81 deletions.
diff --git a/pkg/logs/events.go b/pkg/logs/events.go
@@ -262,9 +262,7 @@ func (ls *LogsService) stopConsumer(ctx context.Context, wg *sync.WaitGroup, con
  err := adapter.Stop(id)
  if err != nil {
  l.Errorw("stop error", "adapter", adapter.Name(), "error", err)
- continue
  }
-
  return
  }
 

diff --git a/pkg/logs/events_test.go b/pkg/logs/events_test.go
@@ -90,7 +90,7 @@ func TestLogs_EventsFlow(t *testing.T) {
  assert.NoError(t, err)
 
  // cooldown stop time
- time.Sleep(waitTime * 2)
+ time.Sleep(waitTime)
 
  // then all adapters should be gracefully stopped
  assert.Equal(t, 0, log.GetConsumersStats(ctx).Count)
@@ -316,7 +316,7 @@ func TestLogs_EventsFlow(t *testing.T) {
  assert.Equal(t, "stop-queued", string(r.Message))
 
  // there will be wait for mess
- time.Sleep(waitTime * 2)
+ time.Sleep(waitTime)
 
  // then all adapters should be gracefully stopped
  assert.Equal(t, 0, log.GetConsumersStats(ctx).Count)

diff --git a/pkg/logs/pb/logs.pb.go b/pkg/logs/pb/logs.pb.go