chore(blooms): Remove ID field from task struct (#12851)

We've seen a few cases where creating the ULID failed for unknown reasons, and the ID is not really used. It was only useful early on in the development for debugging.

Signed-off-by: Christian Haudum <[email protected]>

go.mod

@@ -75,7 +75,7 @@ require (
 	github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f
 	github.com/ncw/swift v1.0.53
 	github.com/oklog/run v1.1.0
-	github.com/oklog/ulid v1.3.1
+	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/opentracing-contrib/go-grpc v0.0.0-20210225150812-73cb765af46e
 	github.com/opentracing-contrib/go-stdlib v1.0.0
 	github.com/opentracing/opentracing-go v1.2.0

pkg/bloomgateway/bloomgateway.go

@@ -278,10 +278,7 @@ func (g *Gateway) FilterChunkRefs(ctx context.Context, req *logproto.FilterChunk
 	tasks := make([]Task, 0, len(seriesByDay))
 	responses := make([][]v1.Output, 0, len(seriesByDay))
 	for _, seriesForDay := range seriesByDay {
-		task, err := NewTask(ctx, tenantID, seriesForDay, filters, blocks)
-		if err != nil {
-			return nil, err
-		}
+		task := newTask(ctx, tenantID, seriesForDay, filters, blocks)
 		// TODO(owen-d): include capacity in constructor?
 		task.responses = responsesPool.Get(len(seriesForDay.series))
 		tasks = append(tasks, task)
@@ -298,7 +295,6 @@ func (g *Gateway) FilterChunkRefs(ctx context.Context, req *logproto.FilterChunk
 	for _, task := range tasks {
 		task := task
 		task.enqueueTime = time.Now()
-		level.Info(logger).Log("msg", "enqueue task", "task", task.ID, "table", task.table, "series", len(task.series))
 
 		// TODO(owen-d): gracefully handle full queues
 		if err := g.queue.Enqueue(tenantID, nil, task, func() {
@@ -329,7 +325,6 @@ func (g *Gateway) FilterChunkRefs(ctx context.Context, req *logproto.FilterChunk
 			stats.Status = "cancel"
 			return nil, errors.Wrap(ctx.Err(), "request failed")
 		case task := <-tasksCh:
-			level.Info(logger).Log("msg", "task done", "task", task.ID, "err", task.Err())
 			if task.Err() != nil {
 				stats.Status = labelFailure
 				return nil, errors.Wrap(task.Err(), "request failed")

pkg/bloomgateway/multiplexing.go

@@ -2,11 +2,9 @@ package bloomgateway
 
 import (
 	"context"
-	"math/rand"
 	"sync"
 	"time"
 
-	"github.com/oklog/ulid"
 	"github.com/prometheus/common/model"
 
 	"github.com/grafana/loki/v3/pkg/logproto"
@@ -20,10 +18,6 @@ const (
 	Day = 24 * time.Hour
 )
 
-var (
-	entropy = rand.New(rand.NewSource(time.Now().UnixNano()))
-)
-
 type tokenSettings struct {
 	nGramLen int
 }
@@ -45,10 +39,8 @@ func (e *wrappedError) Set(err error) {
 
 // Task is the data structure that is enqueued to the internal queue and dequeued by query workers
 type Task struct {
-	// ID is a lexcographically sortable unique identifier of the task
-	ID ulid.ULID
-	// Tenant is the tenant ID
-	Tenant string
+	// tenant is the tenant ID
+	tenant string
 
 	// channel to write partial responses to
 	resCh chan v1.Output
@@ -79,18 +71,9 @@ type Task struct {
 	enqueueTime time.Time
 }
 
-// NewTask returns a new Task that can be enqueued to the task queue.
-// In addition, it returns a result and an error channel, as well
-// as an error if the instantiation fails.
-func NewTask(ctx context.Context, tenantID string, refs seriesWithInterval, filters []syntax.LineFilterExpr, blocks []bloomshipper.BlockRef) (Task, error) {
-	key, err := ulid.New(ulid.Now(), entropy)
-	if err != nil {
-		return Task{}, err
-	}
-
-	task := Task{
-		ID:       key,
-		Tenant:   tenantID,
+func newTask(ctx context.Context, tenantID string, refs seriesWithInterval, filters []syntax.LineFilterExpr, blocks []bloomshipper.BlockRef) Task {
+	return Task{
+		tenant:   tenantID,
 		err:      new(wrappedError),
 		resCh:    make(chan v1.Output),
 		filters:  filters,
@@ -101,7 +84,6 @@ func NewTask(ctx context.Context, tenantID string, refs seriesWithInterval, filt
 		ctx:      ctx,
 		done:     make(chan struct{}),
 	}
-	return task, nil
 }
 
 // Bounds implements Bounded
@@ -130,9 +112,8 @@ func (t Task) CloseWithError(err error) {
 
 // Copy returns a copy of the existing task but with a new slice of grouped chunk refs
 func (t Task) Copy(series []*logproto.GroupedChunkRefs) Task {
-	// do not copy ID to distinguish it as copied task
 	return Task{
-		Tenant:   t.Tenant,
+		tenant:   t.tenant,
 		err:      t.err,
 		resCh:    t.resCh,
 		filters:  t.filters,

pkg/bloomgateway/multiplexing_test.go

@@ -31,8 +31,7 @@ func TestTask(t *testing.T) {
 			},
 		}
 		swb := partitionRequest(req)[0]
-		task, err := NewTask(context.Background(), "tenant", swb, nil, nil)
-		require.NoError(t, err)
+		task := newTask(context.Background(), "tenant", swb, nil, nil)
 		from, through := task.Bounds()
 		require.Equal(t, ts.Add(-1*time.Hour), from)
 		require.Equal(t, ts, through)
@@ -45,8 +44,7 @@ func createTasksForRequests(t *testing.T, tenant string, requests ...*logproto.F
 	tasks := make([]Task, 0, len(requests))
 	for _, r := range requests {
 		for _, swb := range partitionRequest(r) {
-			task, err := NewTask(context.Background(), tenant, swb, nil, nil)
-			require.NoError(t, err)
+			task := newTask(context.Background(), tenant, swb, nil, nil)
 			tasks = append(tasks, task)
 		}
 	}
@@ -63,7 +61,7 @@ func TestTask_RequestIterator(t *testing.T) {
 			interval: bloomshipper.Interval{Start: 0, End: math.MaxInt64},
 			series:   []*logproto.GroupedChunkRefs{},
 		}
-		task, _ := NewTask(context.Background(), tenant, swb, []syntax.LineFilterExpr{}, nil)
+		task := newTask(context.Background(), tenant, swb, []syntax.LineFilterExpr{}, nil)
 		it := task.RequestIter(tokenizer)
 		// nothing to iterate over
 		require.False(t, it.Next())

pkg/bloomgateway/processor.go

@@ -40,7 +40,7 @@ func (p *processor) run(ctx context.Context, tasks []Task) error {
 }
 
 func (p *processor) runWithBounds(ctx context.Context, tasks []Task, bounds v1.MultiFingerprintBounds) error {
-	tenant := tasks[0].Tenant
+	tenant := tasks[0].tenant
 	level.Info(p.logger).Log(
 		"msg", "process tasks with bounds",
 		"tenant", tenant,
@@ -157,7 +157,7 @@ func (p *processor) processBlock(_ context.Context, blockQuerier *v1.BlockQuerie
 	for _, task := range tasks {
 		if sp := opentracing.SpanFromContext(task.ctx); sp != nil {
 			md, _ := blockQuerier.Metadata()
-			blk := bloomshipper.BlockRefFrom(task.Tenant, task.table.String(), md)
+			blk := bloomshipper.BlockRefFrom(task.tenant, task.table.String(), md)
 			sp.LogKV("process block", blk.String(), "series", len(task.series))
 		}
 

pkg/bloomgateway/processor_test.go

@@ -126,7 +126,7 @@ func TestProcessor(t *testing.T) {
 		}
 
 		t.Log("series", len(swb.series))
-		task, _ := NewTask(ctx, "fake", swb, filters, nil)
+		task := newTask(ctx, "fake", swb, filters, nil)
 		tasks := []Task{task}
 
 		results := atomic.NewInt64(0)
@@ -178,7 +178,7 @@ func TestProcessor(t *testing.T) {
 		}
 
 		t.Log("series", len(swb.series))
-		task, _ := NewTask(ctx, "fake", swb, filters, blocks)
+		task := newTask(ctx, "fake", swb, filters, blocks)
 		tasks := []Task{task}
 
 		results := atomic.NewInt64(0)
@@ -227,7 +227,7 @@ func TestProcessor(t *testing.T) {
 		}
 
 		t.Log("series", len(swb.series))
-		task, _ := NewTask(ctx, "fake", swb, filters, nil)
+		task := newTask(ctx, "fake", swb, filters, nil)
 		tasks := []Task{task}
 
 		results := atomic.NewInt64(0)

pkg/bloomgateway/worker.go

@@ -100,7 +100,6 @@ func (w *worker) running(_ context.Context) error {
 				w.queue.ReleaseRequests(items)
 				return errors.Errorf("failed to cast dequeued item to Task: %v", item)
 			}
-			level.Debug(w.logger).Log("msg", "dequeued task", "task", task.ID)
 			_ = w.pending.Dec()
 			w.metrics.queueDuration.WithLabelValues(w.id).Observe(time.Since(task.enqueueTime).Seconds())
 			FromContext(task.ctx).AddQueueTime(time.Since(task.enqueueTime))