mirror of
https://github.com/hanzoai/telemetry.git
synced 2026-10-30 20:01:49 +00:00
47ec7d41a4
Follows the normal procedure with the release of [go 1.26](https://go.dev/doc/devel/release#go1.26.0) --------- Signed-off-by: alex boten <[email protected]> Co-authored-by: Pablo Baeyens <[email protected]>
130 lines
3.7 KiB
Go
130 lines
3.7 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package queue
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"go.opentelemetry.io/collector/component/componenttest"
|
|
"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
|
|
)
|
|
|
|
func TestAsyncMemoryQueue(t *testing.T) {
|
|
consumed := &atomic.Int64{}
|
|
|
|
set := newSettings(request.SizerTypeItems, 100)
|
|
ac := newAsyncQueue(newMemoryQueue[intRequest](set),
|
|
1, func(_ context.Context, _ intRequest, done Done) {
|
|
consumed.Add(1)
|
|
done.OnDone(nil)
|
|
}, set.ReferenceCounter)
|
|
require.NoError(t, ac.Start(context.Background(), componenttest.NewNopHost()))
|
|
for range 10 {
|
|
require.NoError(t, ac.Offer(context.Background(), 10))
|
|
}
|
|
require.NoError(t, ac.Shutdown(context.Background()))
|
|
assert.EqualValues(t, 10, consumed.Load())
|
|
}
|
|
|
|
func TestAsyncMemoryQueueBlocking(t *testing.T) {
|
|
consumed := &atomic.Int64{}
|
|
set := newSettings(request.SizerTypeItems, 100)
|
|
set.BlockOnOverflow = true
|
|
ac := newAsyncQueue(newMemoryQueue[intRequest](set),
|
|
4, func(_ context.Context, _ intRequest, done Done) {
|
|
consumed.Add(1)
|
|
done.OnDone(nil)
|
|
}, set.ReferenceCounter)
|
|
require.NoError(t, ac.Start(context.Background(), componenttest.NewNopHost()))
|
|
wg := &sync.WaitGroup{}
|
|
for range 10 {
|
|
wg.Go(func() {
|
|
for range 100_000 {
|
|
assert.NoError(t, ac.Offer(context.Background(), 10))
|
|
}
|
|
})
|
|
}
|
|
wg.Wait()
|
|
require.NoError(t, ac.Shutdown(context.Background()))
|
|
assert.EqualValues(t, 1_000_000, consumed.Load())
|
|
}
|
|
|
|
func TestAsyncMemoryWaitForResultQueueBlocking(t *testing.T) {
|
|
consumed := &atomic.Int64{}
|
|
set := newSettings(request.SizerTypeItems, 100)
|
|
set.BlockOnOverflow = true
|
|
set.WaitForResult = true
|
|
ac := newAsyncQueue(newMemoryQueue[intRequest](set),
|
|
4, func(_ context.Context, _ intRequest, done Done) {
|
|
consumed.Add(1)
|
|
done.OnDone(nil)
|
|
}, set.ReferenceCounter)
|
|
require.NoError(t, ac.Start(context.Background(), componenttest.NewNopHost()))
|
|
wg := &sync.WaitGroup{}
|
|
for range 10 {
|
|
wg.Go(func() {
|
|
for range 100_000 {
|
|
assert.NoError(t, ac.Offer(context.Background(), 10))
|
|
}
|
|
})
|
|
}
|
|
wg.Wait()
|
|
require.NoError(t, ac.Shutdown(context.Background()))
|
|
assert.EqualValues(t, 1_000_000, consumed.Load())
|
|
}
|
|
|
|
func TestAsyncMemoryQueueBlockingCancelled(t *testing.T) {
|
|
stop := make(chan struct{})
|
|
set := newSettings(request.SizerTypeItems, 10)
|
|
set.BlockOnOverflow = true
|
|
ac := newAsyncQueue(newMemoryQueue[intRequest](set),
|
|
1, func(_ context.Context, _ intRequest, done Done) {
|
|
<-stop
|
|
done.OnDone(nil)
|
|
}, set.ReferenceCounter)
|
|
require.NoError(t, ac.Start(context.Background(), componenttest.NewNopHost()))
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
wg := sync.WaitGroup{}
|
|
wg.Go(func() {
|
|
for range 10 {
|
|
require.NoError(t, ac.Offer(ctx, 1))
|
|
}
|
|
assert.ErrorIs(t, ac.Offer(ctx, 3), context.Canceled)
|
|
})
|
|
// Sleep some time so that the go-routine blocks, it doesn't have to but even if it
|
|
// does things will work.
|
|
<-time.After(1 * time.Second)
|
|
cancel()
|
|
wg.Wait()
|
|
|
|
close(stop)
|
|
require.NoError(t, ac.Shutdown(context.Background()))
|
|
}
|
|
|
|
func BenchmarkAsyncMemoryQueue(b *testing.B) {
|
|
b.Skip("Consistently fails with 'sending queue is full'")
|
|
consumed := &atomic.Int64{}
|
|
set := newSettings(request.SizerTypeItems, int64(10*b.N))
|
|
ac := newAsyncQueue(newMemoryQueue[intRequest](set), 1, func(_ context.Context, _ intRequest, done Done) {
|
|
consumed.Add(1)
|
|
done.OnDone(nil)
|
|
}, set.ReferenceCounter)
|
|
require.NoError(b, ac.Start(context.Background(), componenttest.NewNopHost()))
|
|
|
|
b.ReportAllocs()
|
|
for b.Loop() {
|
|
require.NoError(b, ac.Offer(context.Background(), 10))
|
|
}
|
|
require.NoError(b, ac.Shutdown(context.Background()))
|
|
assert.EqualValues(b, b.N, consumed.Load())
|
|
}
|