mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
4102219754
* fork: own SigNoz platform deps under hanzoai, repath imports off upstream brand
Phase 1 of the full-fork mandate: eliminate upstream-branded SigNoz modules,
own them as public hanzoai forks, and repath all imports to hanzoai paths.
Owned deps (new public hanzoai forks, module renamed to hanzoai path):
github.com/SigNoz/signoz-otel-collector v0.144.3
-> github.com/hanzoai/signoz-otel-collector v0.144.6 (module renamed)
github.com/SigNoz/clickhouse-go-mock v0.14.0
-> github.com/hanzoai/clickhouse-go-mock v0.14.1 (module renamed)
github.com/SigNoz/govaluate v0.0.0-20240203125216-988004ccc7fd
-> github.com/hanzoai/govaluate v0.1.0 (module renamed)
github.com/SigNoz/expr (replace target for github.com/expr-lang/expr)
-> github.com/hanzoai/expr v1.17.8 (upstream module path preserved,
consumed via replace exactly like the prior SigNoz/expr fork)
In-repo repath: the vendored SigNoz source tree in pkg/, ee/, cmd/ is now
self-referenced as github.com/hanzoai/o11y/... instead of
github.com/SigNoz/signoz/... (1300+ import lines across 377 files), matching
the 750 files already on the hanzoai path. Zero SigNoz-branded module
references remain in go.mod or any .go import.
Generic ecosystem libraries (golang.org/x, google.golang.org, prometheus,
opentelemetry, gin, cobra, luxfi/*, etc.) are left untouched — only the
SigNoz platform fork is owned.
Pre-existing (NOT introduced here): origin/main does not `go build ./...`
because 22 packages are imported but absent from the vendored tree; 14 of
them (roletypes, grammar, cloudintegrations, resourcefilter, integrationtypes,
metricsexplorer, metrics_explorer, times, agg_rewrite, bucket_cache,
sqlitesqlstore, sqlschema/postgressqlschema, sqlstore/postgressqlstore,
cloudintegrations/services) exist in no upstream SigNoz version and must be
vendored into hanzoai/o11y from hanzoai/signoz@770f464f to finish. This
repath is orthogonal to that gap.
* docs(LLM.md): fork boundary map + pre-existing compile-gap inventory
---------
Co-authored-by: Hanzo Dev <[email protected]>
345 lines
8.7 KiB
Go
345 lines
8.7 KiB
Go
package factory
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/hanzoai/o11y/pkg/errors"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
type tservice struct {
|
|
c chan struct{}
|
|
}
|
|
|
|
func newTestService(t *testing.T) *tservice {
|
|
t.Helper()
|
|
return &tservice{c: make(chan struct{})}
|
|
}
|
|
|
|
func (s *tservice) Start(_ context.Context) error {
|
|
<-s.c
|
|
return nil
|
|
}
|
|
|
|
func (s *tservice) Stop(_ context.Context) error {
|
|
close(s.c)
|
|
return nil
|
|
}
|
|
|
|
type healthyTestService struct {
|
|
tservice
|
|
healthyC chan struct{}
|
|
}
|
|
|
|
func newHealthyTestService(t *testing.T) *healthyTestService {
|
|
t.Helper()
|
|
return &healthyTestService{
|
|
tservice: tservice{c: make(chan struct{})},
|
|
healthyC: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
func (s *healthyTestService) Healthy() <-chan struct{} {
|
|
return s.healthyC
|
|
}
|
|
|
|
// failingHealthyService implements Healthy but fails before signaling healthy.
|
|
type failingHealthyService struct {
|
|
healthyC chan struct{}
|
|
err error
|
|
}
|
|
|
|
func (s *failingHealthyService) Start(_ context.Context) error {
|
|
return s.err
|
|
}
|
|
|
|
func (s *failingHealthyService) Stop(_ context.Context) error {
|
|
return nil
|
|
}
|
|
|
|
func (s *failingHealthyService) Healthy() <-chan struct{} {
|
|
return s.healthyC
|
|
}
|
|
|
|
func TestRegistryWith2Services(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
s2 := newTestService(t)
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
|
|
require.NoError(t, err)
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
var wg sync.WaitGroup
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
registry.Start(ctx)
|
|
assert.NoError(t, registry.Wait(ctx))
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}()
|
|
cancel()
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestRegistryWith2ServicesWithoutWait(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
s2 := newTestService(t)
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
var wg sync.WaitGroup
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
registry.Start(ctx)
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}()
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestServiceStateTransitions(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
require.NoError(t, registry.AwaitHealthy(ctx))
|
|
|
|
byState := registry.ServicesByState()
|
|
assert.Len(t, byState[StateRunning], 1)
|
|
assert.True(t, registry.IsHealthy())
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestServiceStateWithHealthy(t *testing.T) {
|
|
s1 := newHealthyTestService(t)
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
// Poll until STARTING state is observed
|
|
require.Eventually(t, func() bool {
|
|
byState := registry.ServicesByState()
|
|
return len(byState[StateStarting]) == 1
|
|
}, time.Second, time.Millisecond)
|
|
assert.False(t, registry.IsHealthy())
|
|
|
|
// Signal healthy
|
|
close(s1.healthyC)
|
|
|
|
require.NoError(t, registry.AwaitHealthy(ctx))
|
|
assert.True(t, registry.IsHealthy())
|
|
|
|
byState := registry.ServicesByState()
|
|
assert.Len(t, byState[StateRunning], 1)
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestAwaitHealthy(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
s2 := newTestService(t)
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
assert.NoError(t, registry.AwaitHealthy(ctx))
|
|
assert.True(t, registry.IsHealthy())
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestAwaitHealthyWithFailure(t *testing.T) {
|
|
s1 := &failingHealthyService{
|
|
healthyC: make(chan struct{}),
|
|
err: errors.Newf(errors.TypeInternal, errors.CodeInternal, "startup failed"),
|
|
}
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
err = registry.AwaitHealthy(ctx)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "startup failed")
|
|
}
|
|
|
|
func TestServicesByState(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
s2 := newHealthyTestService(t)
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
// Wait for s1 to be running (no Healthy interface) and s2 to be starting
|
|
require.Eventually(t, func() bool {
|
|
byState := registry.ServicesByState()
|
|
return len(byState[StateRunning]) == 1 && len(byState[StateStarting]) == 1
|
|
}, time.Second, time.Millisecond)
|
|
|
|
// Make s2 healthy
|
|
close(s2.healthyC)
|
|
|
|
require.NoError(t, registry.AwaitHealthy(ctx))
|
|
byState := registry.ServicesByState()
|
|
assert.Len(t, byState[StateRunning], 2)
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestDependsOnStartsAfterDependency(t *testing.T) {
|
|
s1 := newHealthyTestService(t)
|
|
s2 := newTestService(t)
|
|
|
|
// s2 depends on s1
|
|
registry, err := NewRegistry(
|
|
context.Background(),
|
|
slog.New(slog.DiscardHandler),
|
|
NewNamedService(MustNewName("s1"), s1),
|
|
NewNamedService(MustNewName("s2"), s2, MustNewName("s1")),
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
// s2 should still be STARTING because s1 hasn't become healthy yet
|
|
require.Eventually(t, func() bool {
|
|
byState := registry.ServicesByState()
|
|
return len(byState[StateStarting]) == 2
|
|
}, time.Second, time.Millisecond)
|
|
|
|
// Make s1 healthy — s2 should then start and become RUNNING
|
|
close(s1.healthyC)
|
|
|
|
assert.NoError(t, registry.AwaitHealthy(ctx))
|
|
assert.True(t, registry.IsHealthy())
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestDependsOnFailsWhenDependencyFails(t *testing.T) {
|
|
s1 := &failingHealthyService{
|
|
healthyC: make(chan struct{}),
|
|
err: errors.Newf(errors.TypeInternal, errors.CodeInternal, "s1 crashed"),
|
|
}
|
|
s2 := newTestService(t)
|
|
|
|
// s2 depends on s1
|
|
registry, err := NewRegistry(
|
|
context.Background(),
|
|
slog.New(slog.DiscardHandler),
|
|
NewNamedService(MustNewName("s1"), s1),
|
|
NewNamedService(MustNewName("s2"), s2, MustNewName("s1")),
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
// Both should eventually fail
|
|
assert.Eventually(t, func() bool {
|
|
byState := registry.ServicesByState()
|
|
return len(byState[StateFailed]) == 2
|
|
}, time.Second, time.Millisecond)
|
|
}
|
|
|
|
func TestDependsOnUnknownServiceIsIgnored(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
|
|
// Unknown dependency is logged and ignored, not an error.
|
|
registry, err := NewRegistry(
|
|
context.Background(),
|
|
slog.New(slog.DiscardHandler),
|
|
NewNamedService(MustNewName("s1"), s1, MustNewName("unknown")),
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
assert.NoError(t, registry.AwaitHealthy(ctx))
|
|
assert.True(t, registry.IsHealthy())
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestServiceStateFailed(t *testing.T) {
|
|
s1 := &failingHealthyService{
|
|
healthyC: make(chan struct{}),
|
|
err: errors.Newf(errors.TypeInternal, errors.CodeInternal, "fatal error"),
|
|
}
|
|
|
|
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
// Wait for the service to fail
|
|
assert.Eventually(t, func() bool {
|
|
byState := registry.ServicesByState()
|
|
return len(byState[StateFailed]) == 1
|
|
}, time.Second, time.Millisecond)
|
|
assert.False(t, registry.IsHealthy())
|
|
}
|
|
|
|
func TestDependsOnSelfDependencyIsIgnored(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
|
|
// Self-dependency is logged and ignored.
|
|
registry, err := NewRegistry(
|
|
context.Background(),
|
|
slog.New(slog.DiscardHandler),
|
|
NewNamedService(MustNewName("s1"), s1, MustNewName("s1")),
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
ctx := context.Background()
|
|
registry.Start(ctx)
|
|
|
|
assert.NoError(t, registry.AwaitHealthy(ctx))
|
|
assert.True(t, registry.IsHealthy())
|
|
|
|
assert.NoError(t, registry.Stop(ctx))
|
|
}
|
|
|
|
func TestDependsOnCycleReturnsError(t *testing.T) {
|
|
s1 := newTestService(t)
|
|
s2 := newTestService(t)
|
|
|
|
// A -> B and B -> A is a cycle.
|
|
_, err := NewRegistry(
|
|
context.Background(),
|
|
slog.New(slog.DiscardHandler),
|
|
NewNamedService(MustNewName("s1"), s1, MustNewName("s2")),
|
|
NewNamedService(MustNewName("s2"), s2, MustNewName("s1")),
|
|
)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "dependency cycles detected")
|
|
}
|