From f4225d96460285293fa698e6ce4a8c9894a1a9a1 Mon Sep 17 00:00:00 2001 From: "timmy.yuan" Date: Tue, 13 Oct 2026 17:39:09 +0800 Subject: [PATCH] test(cpdaemon,cpgolib): check cpm config mapping, buffer sizing, sync state and handshake gremlins left many mutants alive in cpdaemon/pkg/cpm and cpgolib/cpworker (#243 item 6): branches the tests run but whose result they never check. New and extended tests: - worker_task_builder: the VXLAN tag format follows apiVersion (vni1 with the service tag or 0xFFFFFF; vni2 with domain/point/extension flag, VXLAN-WIRE-FORMAT.md sections 5, 8, 9); rate limit 0 left unset; vxlan port, zmq service tag and file dump interval reach the output; packetSplitBytes 0 accepted; --nofilter without bind_device rejected. - worker_mgr: auto_nic_buffer splits the memory limit over the tasks, CPM memLimit overrides the default only when > 0, too little memory is an error; the pipeline section gets the memory left after the libpcap buffers but at least MinBufferSizeMb; cpu_affinity passed through; the control socket directory is created before the worker starts. - syncer: register sends start seconds plus microseconds; a new daemon id stops the worker and resets the sync state; a failing strategy sync is retried SyncStrategyMaxRetries times; metrics sampling microseconds stay below 1e6; the uuid file is reused and an empty one refilled; IncludingNICs limits the registered interfaces. - client: SetDefaults keeps set values; Register sends labels and networkInterfaces as [] and fails on a non-2xx status or a body code >= 400. models: HasInstances. utils: splitArgs drops empty arguments. synclog: level threshold, entry microseconds. - cpgolib: Dial fails when the handshake reply is not OK or not JSON; a timeout of 0 falls back to DefaultTimeout. gremlins test efficacy (v0.5.0, before -> after): - cpdaemon/pkg/cpm: 63.4% -> 83.3%. The rest are error branches that only log, CPU/memory figures read from the live process, log-only formatting, and equivalent mutants. - cpgolib/cpworker: 76.0% -> 85.7%. The rest are equivalent mutants and the omitted empty "arguments" field, which cpworker never reads. Refs #243 --- cpdaemon/pkg/cpm/client_test.go | 78 ++++++++++ cpdaemon/pkg/cpm/models_test.go | 19 +++ cpdaemon/pkg/cpm/syncer_test.go | 135 +++++++++++++++++- cpdaemon/pkg/cpm/synclog_test.go | 21 +++ cpdaemon/pkg/cpm/utils_test.go | 33 +++++ cpdaemon/pkg/cpm/worker_mgr_test.go | 111 ++++++++++++++ .../cpm/worker_task_builder_output_test.go | 108 ++++++++++++++ .../cpm/worker_task_builder_validate_test.go | 7 + cpgolib/cpworker/unix_client_test.go | 50 ++++++- 9 files changed, 559 insertions(+), 3 deletions(-) create mode 100644 cpdaemon/pkg/cpm/models_test.go create mode 100644 cpdaemon/pkg/cpm/utils_test.go create mode 100644 cpdaemon/pkg/cpm/worker_task_builder_output_test.go diff --git a/cpdaemon/pkg/cpm/client_test.go b/cpdaemon/pkg/cpm/client_test.go index 0bf512bd..8dbfc6bc 100644 --- a/cpdaemon/pkg/cpm/client_test.go +++ b/cpdaemon/pkg/cpm/client_test.go @@ -2,6 +2,7 @@ package cpm import ( "context" + "encoding/json" "net/http" "net/http/httptest" "testing" @@ -52,3 +53,80 @@ func TestHttpClient_HonoursContextCancellation(t *testing.T) { }) } } + +// Zero values get the defaults; values the caller set are kept. +func TestClientConfig_SetDefaults(t *testing.T) { + var zero ClientConfig + zero.SetDefaults() + assert.Equal(t, ClientConfig{ + Timeout: 15 * time.Second, + DialTimeout: 5 * time.Second, + ResponseHeaderTimeout: 15 * time.Second, + MaxIdleConns: 10, + MaxIdleConnsPerHost: 5, + }, zero) + + set := ClientConfig{ + Timeout: time.Second, + DialTimeout: 2 * time.Second, + ResponseHeaderTimeout: 3 * time.Second, + MaxIdleConns: 4, + MaxIdleConnsPerHost: 6, + } + got := set + got.SetDefaults() + assert.Equal(t, set, got) +} + +// CPM requires labels and networkInterfaces to be arrays, never null (RegisterRequest). +func TestHttpClient_RegisterSendsEmptyArrays(t *testing.T) { + var body map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewDecoder(r.Body).Decode(&body) + _, _ = w.Write([]byte(`{"id":1}`)) + })) + t.Cleanup(srv.Close) + + client, err := NewHttpClient(srv.URL, ClientConfig{}) + require.NoError(t, err) + _, err = client.Register(context.Background(), RegisterRequest{}) + require.NoError(t, err) + assert.Equal(t, []any{}, body["labels"]) + assert.Equal(t, []any{}, body["networkInterfaces"]) +} + +// CPM reports a failure either with a non-2xx status or with a body code of 400 or more. +func TestHttpClient_RegisterErrorResponses(t *testing.T) { + cases := []struct { + name string + status int + body string + wantErr bool + }{ + {"ok", http.StatusOK, `{"id":1}`, false}, + {"body_code_399", http.StatusOK, `{"code":399,"id":1}`, false}, + {"body_code_400", http.StatusOK, `{"code":400,"msg":"bad request"}`, true}, + {"status_299", 299, `{"id":1}`, false}, + {"status_300", http.StatusMultipleChoices, `{"id":1}`, true}, + {"status_500", http.StatusInternalServerError, `{}`, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.status) + _, _ = w.Write([]byte(tc.body)) + })) + t.Cleanup(srv.Close) + + client, err := NewHttpClient(srv.URL, ClientConfig{}) + require.NoError(t, err) + res, err := client.Register(context.Background(), RegisterRequest{}) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, int64(1), res.Id) + }) + } +} diff --git a/cpdaemon/pkg/cpm/models_test.go b/cpdaemon/pkg/cpm/models_test.go new file mode 100644 index 00000000..4d721c03 --- /dev/null +++ b/cpdaemon/pkg/cpm/models_test.go @@ -0,0 +1,19 @@ +package cpm + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +// Only a strategy that names KVM instances needs the active instance list. +func TestSyncStrategyResponse_HasInstances(t *testing.T) { + assert.False(t, (&SyncStrategyResponse{}).HasInstances()) + assert.False(t, (&SyncStrategyResponse{Strategy: []StrategyEntry{ + {InterfaceNames: []string{"eth0"}, InstanceNames: []string{}}, + }}).HasInstances()) + assert.True(t, (&SyncStrategyResponse{Strategy: []StrategyEntry{ + {InterfaceNames: []string{"eth0"}}, + {InstanceNames: []string{"vm1"}}, + }}).HasInstances()) +} diff --git a/cpdaemon/pkg/cpm/syncer_test.go b/cpdaemon/pkg/cpm/syncer_test.go index 4f0cfc27..b9862804 100644 --- a/cpdaemon/pkg/cpm/syncer_test.go +++ b/cpdaemon/pkg/cpm/syncer_test.go @@ -3,8 +3,11 @@ package cpm import ( "context" "encoding/json" + "errors" "fmt" "log/slog" + "os" + "path/filepath" "slices" "sync" "testing" @@ -119,6 +122,7 @@ func (r *testWorkerManagerRecoder) SetLogger(lg *slog.Logger) { type ( testRegisterFn func(ctx context.Context, req RegisterRequest) (*RegisterResponse, error) testSyncStrategyFn func(ctx context.Context, daemonId int64, version int32) (*SyncStrategyResult, error) + testSyncMetricsFn func(ctx context.Context, daemonId int64, req SyncMetricsRequest) error ) type testClient struct { @@ -126,6 +130,7 @@ type testClient struct { registerFn testRegisterFn syncStrategyFn testSyncStrategyFn + syncMetricsFn testSyncMetricsFn } func (c *testClient) Register(ctx context.Context, req RegisterRequest) (*RegisterResponse, error) { @@ -143,7 +148,10 @@ func (c *testClient) SyncStrategy(ctx context.Context, daemonId int64, version i func (c *testClient) SyncMetrics(ctx context.Context, daemonId int64, req SyncMetricsRequest) error { c.mu.Lock() defer c.mu.Unlock() - return nil + if c.syncMetricsFn == nil { + return nil + } + return c.syncMetricsFn(ctx, daemonId, req) } func TestSyncer(t *testing.T) { @@ -300,3 +308,128 @@ func TestNetworkInterfaces(t *testing.T) { require.NoError(t, err) fmt.Println(string(data)) } + +// newTestSyncer builds a syncer with its uuid file in a temporary directory. +func newTestSyncer(t *testing.T, client Client, workerMgr IWorkerManager, cfg SyncerConfig) *Syncer { + t.Helper() + cfg.RegCfg.PlatformId = "test" + if cfg.RegCfg.UuidFile == "" { + cfg.RegCfg.UuidFile = filepath.Join(t.TempDir(), "uuid") + } + s, err := newSyncer(client, workerMgr, testTool{}, cfg) + require.NoError(t, err) + return s +} + +// CPM receives timestamps as whole seconds plus the microseconds within that second. +func TestSyncer_registerSendsStartTime(t *testing.T) { + var got RegisterRequest + client := &testClient{registerFn: func(ctx context.Context, req RegisterRequest) (*RegisterResponse, error) { + got = req + return &RegisterResponse{Id: 1}, nil + }} + s := newTestSyncer(t, client, &testWorkerManagerRecoder{}, SyncerConfig{}) + s.startTime = time.Unix(1700000000, 123456789) + + require.NoError(t, s.doRegister(context.Background())) + assert.Equal(t, int64(1700000000), got.StartTimestamp) + assert.Equal(t, int64(123456), got.StartMicroTimestamp) +} + +// A new daemon id from CPM means the strategy synced so far belongs to another registration: +// the worker is stopped and the sync state reset. The same id keeps both. +func TestSyncer_registerDaemonIdChange(t *testing.T) { + for _, tc := range []struct { + name string + regId int64 + wantStop bool + }{ + {"same_id", 1, false}, + {"new_id", 2, true}, + } { + t.Run(tc.name, func(t *testing.T) { + client := &testClient{registerFn: func(ctx context.Context, req RegisterRequest) (*RegisterResponse, error) { + return &RegisterResponse{Id: tc.regId}, nil + }} + workerMgr := &testWorkerManagerRecoder{} + s := newTestSyncer(t, client, workerMgr, SyncerConfig{}) + s.syncResp = &SyncStrategyResponse{DaemonId: 1} + + require.NoError(t, s.doRegister(context.Background())) + assert.Equal(t, tc.wantStop, slices.Contains(workerMgr.Calls(), "Stop")) + assert.Equal(t, tc.wantStop, s.syncResp == nil) + }) + } +} + +// syncLoop retries a failed strategy sync SyncStrategyMaxRetries times, then returns the error. +func TestSyncer_syncStrategyRetries(t *testing.T) { + var calls int + client := &testClient{syncStrategyFn: func(ctx context.Context, daemonId int64, version int32) (*SyncStrategyResult, error) { + calls++ + return nil, errors.New("cpm unreachable") + }} + s := newTestSyncer(t, client, &testWorkerManagerRecoder{}, SyncerConfig{ + SyncStrategyInterval: time.Millisecond, + SyncStrategyMaxRetries: 2, + SyncMetricInterval: time.Hour, + }) + s.regResp = &RegisterResponse{Id: 1} + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.ErrorContains(t, s.syncLoop(ctx), "cpm unreachable") + assert.Equal(t, 3, calls, "one attempt plus SyncStrategyMaxRetries retries") +} + +func TestSyncer_metricsSamplingTime(t *testing.T) { + var got SyncMetricsRequest + client := &testClient{syncMetricsFn: func(ctx context.Context, daemonId int64, req SyncMetricsRequest) error { + got = req + return nil + }} + workerMgr := &testWorkerManagerRecoder{isAlive: true, pid: os.Getpid()} + s := newTestSyncer(t, client, workerMgr, SyncerConfig{}) + s.regResp = &RegisterResponse{Id: 1} + + before := time.Now().Unix() + s.syncMetric(context.Background()) + require.NotNil(t, got.Metrics) + assert.GreaterOrEqual(t, got.Metrics.SamplingTimestamp, before) + assert.GreaterOrEqual(t, got.Metrics.SamplingMicroTimestamp, int64(0)) + assert.Less(t, got.Metrics.SamplingMicroTimestamp, int64(1e6)) + assert.Equal(t, lo.ToPtr(int32(os.Getpid())), got.Pid) +} + +// The daemon keeps the uuid in its uuid file across restarts; an empty file gets a new one. +func TestSyncer_uuidFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "uuid") + require.NoError(t, os.WriteFile(path, []byte("kept-uuid\n"), 0o644)) + s := newTestSyncer(t, &testClient{}, &testWorkerManagerRecoder{}, SyncerConfig{RegCfg: RegConfig{UuidFile: path}}) + assert.Equal(t, "kept-uuid", s.daemonUUID) + + require.NoError(t, os.WriteFile(path, nil, 0o644)) + s = newTestSyncer(t, &testClient{}, &testWorkerManagerRecoder{}, SyncerConfig{RegCfg: RegConfig{UuidFile: path}}) + require.NotEmpty(t, s.daemonUUID) + saved, err := os.ReadFile(path) + require.NoError(t, err) + assert.Equal(t, s.daemonUUID, string(saved)) +} + +// IncludingNICs limits the interfaces registered with CPM; empty registers all of them. +func TestSyncer_includingNICs(t *testing.T) { + all, err := networkInterfaces() + require.NoError(t, err) + if len(all) < 2 { + t.Skip("needs at least two network interfaces") + } + + s := newTestSyncer(t, &testClient{}, &testWorkerManagerRecoder{}, SyncerConfig{}) + assert.Len(t, s.regNics, len(all)) + + s = newTestSyncer(t, &testClient{}, &testWorkerManagerRecoder{}, SyncerConfig{ + RegCfg: RegConfig{IncludingNICs: []string{all[1].Name}}, + }) + require.Len(t, s.regNics, 1) + assert.Equal(t, all[1].Name, s.regNics[0].Name) +} diff --git a/cpdaemon/pkg/cpm/synclog_test.go b/cpdaemon/pkg/cpm/synclog_test.go index ea949026..a1575ae7 100644 --- a/cpdaemon/pkg/cpm/synclog_test.go +++ b/cpdaemon/pkg/cpm/synclog_test.go @@ -1,9 +1,11 @@ package cpm import ( + "context" "errors" "log/slog" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -76,3 +78,22 @@ func TestSyncLog_GroupedAttributesAreQualified(t *testing.T) { assert.Equal(t, "msg=started", syncLogDetails(t, buf)) }) } + +// A record at exactly the handler's level is kept; a lower one is not. +func TestSyncLog_LevelThreshold(t *testing.T) { + h := &SyncLogHandler{w: &SyncLogBuffer{}, level: slog.LevelInfo} + assert.True(t, h.Enabled(context.Background(), slog.LevelInfo)) + assert.True(t, h.Enabled(context.Background(), slog.LevelWarn)) + assert.False(t, h.Enabled(context.Background(), slog.LevelDebug)) +} + +// CPM receives log times as whole seconds plus the microseconds within that second. +func TestSyncLog_EntryTimestamp(t *testing.T) { + buf := &SyncLogBuffer{} + ts := time.Unix(1700000000, 123456789) + require.NoError(t, buf.Write(context.Background(), ts, slog.LevelInfo, "msg")) + entries := buf.Clear() + require.Len(t, entries, 1) + assert.Equal(t, int64(1700000000), entries[0].Timestamp) + assert.Equal(t, int64(123456), entries[0].MicroTimestamp) +} diff --git a/cpdaemon/pkg/cpm/utils_test.go b/cpdaemon/pkg/cpm/utils_test.go new file mode 100644 index 00000000..6e19101e --- /dev/null +++ b/cpdaemon/pkg/cpm/utils_test.go @@ -0,0 +1,33 @@ +package cpm + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// splitArgs splits a strategy's startup string like a shell: runs of spaces separate +// arguments, quotes group them. +func Test_splitArgs(t *testing.T) { + cases := []struct { + input string + want []string + }{ + {"", nil}, + {" ", nil}, + {"-s 65535 -t 0", []string{"-s", "65535", "-t", "0"}}, + {" -s 65535 ", []string{"-s", "65535"}}, + {`-B "eth 1" -M 'do'`, []string{"-B", "eth 1", "-M", "do"}}, + } + for _, tc := range cases { + t.Run(tc.input, func(t *testing.T) { + got, err := splitArgs(tc.input) + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } + + _, err := splitArgs(`-B "eth1`) + assert.ErrorContains(t, err, "unclosed quote") +} diff --git a/cpdaemon/pkg/cpm/worker_mgr_test.go b/cpdaemon/pkg/cpm/worker_mgr_test.go index db97d930..5b0c2093 100644 --- a/cpdaemon/pkg/cpm/worker_mgr_test.go +++ b/cpdaemon/pkg/cpm/worker_mgr_test.go @@ -3,8 +3,10 @@ package cpm import ( "context" "os" + "path/filepath" "testing" + "github.com/samber/lo" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -97,3 +99,112 @@ func TestWorkerManager(t *testing.T) { require.NotZero(t, pid) } } + +// auto_nic_buffer splits the memory limit evenly over the tasks; the limit from CPM replaces +// the configured default only when it is set and positive. +func TestWorkerManager_taskBufferSize(t *testing.T) { + twoNics := &SyncStrategyResponse{Strategy: []StrategyEntry{{InterfaceNames: []string{"eth0", "eth1"}}}} + withLimit := func(mb int64) *SyncStrategyResponse { + r := *twoNics + r.MemLimit = &mb + return &r + } + cases := []struct { + name string + policy string + res *SyncStrategyResponse + want uint64 + wantErr bool + }{ + {"fixed", MemoryPolicyFixedNicBuffer, twoNics, 4, false}, + {"auto_default_limit", MemoryPolicyAutoNicBuffer, twoNics, 128, false}, + {"auto_cpm_limit", MemoryPolicyAutoNicBuffer, withLimit(1000), 500, false}, + {"auto_cpm_limit_0_uses_default", MemoryPolicyAutoNicBuffer, withLimit(0), 128, false}, + {"auto_one_mb_per_task", MemoryPolicyAutoNicBuffer, withLimit(2), 1, false}, + {"auto_limit_below_task_count", MemoryPolicyAutoNicBuffer, withLimit(1), 0, true}, + {"auto_no_tasks", MemoryPolicyAutoNicBuffer, &SyncStrategyResponse{}, 0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + mgr := NewWorkerManager(WorkerConfig{Memory: MemoryConfig{ + Policy: tc.policy, + DefaultLimitMb: 256, + Libpcap: LibpcapMemConfig{FixedBufferSizeMb: 4}, + }}, testTool{}) + got, err := mgr.getTaskBufferSizeMb(tc.res, nil) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// Only the pipeline execution model gets a pipeline section: the memory left after the +// libpcap buffers, but at least MinBufferSizeMb. cpu_affinity is passed through when configured. +func TestWorkerManager_newWorkerConfig(t *testing.T) { + tasks := []*worker.TaskConfig{{Capturer: worker.CapturerConfig{ + Type: worker.CapturerType_Libpcap, + Libpcap: &worker.LibpcapConfig{BufferSizeMB: lo.ToPtr[uint64](100)}, + }}} + + rtc := NewWorkerManager(WorkerConfig{ExecutionModel: worker.EXECUTION_MODEL_RTC}, testTool{}) + cfg := rtc.newWorkerConfig(tasks, &SyncStrategyResponse{}) + assert.Nil(t, cfg.Pipeline) + assert.Nil(t, cfg.CpuAffinity) + + cases := []struct { + name string + cpmLimit *int64 + want uint64 + }{ + {"default_limit", nil, 1024 - 100}, + {"cpm_limit", lo.ToPtr[int64](2048), 2048 - 100}, + {"cpm_limit_0_uses_default", lo.ToPtr[int64](0), 1024 - 100}, + {"left_below_minimum", lo.ToPtr[int64](150), 64}, + {"nothing_left", lo.ToPtr[int64](80), 64}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pipeline := NewWorkerManager(WorkerConfig{ + ExecutionModel: worker.EXECUTION_MODEL_PIPELINE, + CpuAffinity: "1-3", + Pipeline: PipelineConfig{MinBufferSizeMb: 64}, + Memory: MemoryConfig{DefaultLimitMb: 1024}, + }, testTool{}) + cfg := pipeline.newWorkerConfig(tasks, &SyncStrategyResponse{MemLimit: tc.cpmLimit}) + require.NotNil(t, cfg.Pipeline) + assert.Equal(t, tc.want, cfg.Pipeline.BufferSizeMB) + assert.Equal(t, lo.ToPtr("1-3"), cfg.CpuAffinity) + }) + } +} + +// The worker binds its control socket itself, so its directory must exist before the start. +func TestWorkerManager_createsControlSocketDir(t *testing.T) { + sockDir := filepath.Join(t.TempDir(), "run") + mgr := NewWorkerManager( + WorkerConfig{ + PidFile: filepath.Join(t.TempDir(), "w.pid"), + ConfigFile: filepath.Join(t.TempDir(), "w.json"), + Executable: "../worker/fakeworker", + ExecutionModel: worker.EXECUTION_MODEL_RTC, + Control: worker.ControlConfig{ + Type: "unix", + Unix: &worker.ControlUnixConfig{Path: filepath.Join(sockDir, "w.sock")}, + }, + UpdatePolicy: UpdatePolicyRestart, + Memory: MemoryConfig{Policy: MemoryPolicyAutoNicBuffer, DefaultLimitMb: 256}, + }, + testTool{}, + ) + t.Cleanup(func() { mgr.Stop() }) + + var res SyncStrategyResponse + require.NoError(t, testutils.LoadResultFromJSON("testdata/syncStrategy1.json", &res)) + _, err := mgr.CreateIfDead(context.Background(), &res, "", []string{}) + require.NoError(t, err) + assert.DirExists(t, sockDir) +} diff --git a/cpdaemon/pkg/cpm/worker_task_builder_output_test.go b/cpdaemon/pkg/cpm/worker_task_builder_output_test.go new file mode 100644 index 00000000..5f7c9f2f --- /dev/null +++ b/cpdaemon/pkg/cpm/worker_task_builder_output_test.go @@ -0,0 +1,108 @@ +package cpm + +import ( + "testing" + + "github.com/samber/lo" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Netis/cloud-probe/cpdaemon/pkg/worker" +) + +// buildOutput runs one valid strategy through the builder and returns the output of its task. +func buildOutput(t *testing.T, s StrategyEntry) worker.OutputConfig { + t.Helper() + tb := &workerTaskBuilder{tool: testTool{}, buffSize: 256} + tb.addStrategy(s) + tasks, warnings := tb.build() + require.Empty(t, warnings) + require.Len(t, tasks, 1) + require.Len(t, tasks[0].Outputs, 1) + return tasks[0].Outputs[0] +} + +// The strategy fields that reach the cpworker output config (docs/USAGE-CPWORKER.md). +func Test_workerTaskBuilder_outputFields(t *testing.T) { + t.Run("forward_rate_limit", func(t *testing.T) { + s := validStrategy(t, PacketChannelType_GRE) + s.ForwardRateLimit = lo.ToPtr[int32](10) + assert.Equal(t, lo.ToPtr[uint64](10), buildOutput(t, s).RateLimitMbps) + + // 0 means no limit: cpworker's rate_limit_mbps is left unset. + s.ForwardRateLimit = lo.ToPtr[int32](0) + assert.Nil(t, buildOutput(t, s).RateLimitMbps) + }) + + t.Run("vxlan_port", func(t *testing.T) { + s := validStrategy(t, PacketChannelType_VXLAN) + s.Port = lo.ToPtr[int32](4790) + assert.Equal(t, lo.ToPtr[int32](4790), buildOutput(t, s).Vxlan.Port) + }) + + t.Run("zmq_service_tag", func(t *testing.T) { + s := validStrategy(t, PacketChannelType_ZMQ) + s.HasServiceTag = true + s.ServiceTag = lo.ToPtr[int32](7) + assert.Equal(t, lo.ToPtr[uint32](7), buildOutput(t, s).Zmq.ServiceTag) + }) + + t.Run("file_dump_interval", func(t *testing.T) { + s := validStrategy(t, PacketChannelType_FILE) + s.DumpInterval = lo.ToPtr[int32](60) + assert.Equal(t, lo.ToPtr[int32](60), buildOutput(t, s).RotatingFile.MaxFileInterval) + }) +} + +// The VXLAN tag format follows apiVersion (docs/VXLAN-WIRE-FORMAT.md §8): unset or "v1" sends +// vni1, the service tag or 0xFFFFFF without one; any other version sends vni2. +func Test_workerTaskBuilder_vxlanTagFormat(t *testing.T) { + // vni2 with no observation tag: domain 1 (bits 8-31) and point 1 (bits 2-6), §5. + const defaultVni2 = 1<<8 | 1<<2 + + cases := []struct { + name string + apiVersion *string + serviceTag *int32 + vni1 *uint32 + vni2 *uint32 + }{ + {"unset_with_tag", nil, lo.ToPtr[int32](0x1A2), lo.ToPtr[uint32](0x1A2), nil}, + {"unset_without_tag", nil, nil, lo.ToPtr[uint32](0xFFFFFF), nil}, + {"v1_with_tag", lo.ToPtr("v1"), lo.ToPtr[int32](0x1A2), lo.ToPtr[uint32](0x1A2), nil}, + {"v2", lo.ToPtr("v2"), lo.ToPtr[int32](0x1A2), nil, lo.ToPtr[uint32](defaultVni2)}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s := validStrategy(t, PacketChannelType_VXLAN) + s.ApiVersion = tc.apiVersion + s.HasServiceTag = tc.serviceTag != nil + s.ServiceTag = tc.serviceTag + out := buildOutput(t, s) + assert.Equal(t, tc.vni1, out.Vxlan.Vni1) + assert.Equal(t, tc.vni2, out.Vxlan.Vni2) + }) + } +} + +// vni2 carries the observation domain and point of the interface and the extension flag. +// Domain 3568 and point 9 are the vni2 example of docs/VXLAN-WIRE-FORMAT.md §9 (0xDF024); +// the extension flag sets bit 7 (§5). +func Test_workerTaskBuilder_vni2Tag(t *testing.T) { + s := validStrategy(t, PacketChannelType_VXLAN) + s.ApiVersion = lo.ToPtr("v2") + s.HasObservationTag = true + s.ObservationDomainIds = []uint32{3568} + s.ObservationPointIds = []uint8{9} + assert.Equal(t, lo.ToPtr[uint32](0xDF024), buildOutput(t, s).Vxlan.Vni2) + + s.HasExtensionFlag = true + s.ExtensionFlag = lo.ToPtr[int8](1) + assert.Equal(t, lo.ToPtr[uint32](0xDF0A4), buildOutput(t, s).Vxlan.Vni2) + + // An interface without an id of its own gets domain 1 and point 1. + s.HasExtensionFlag = false + s.ObservationDomainIds = nil + s.ObservationPointIds = nil + assert.Equal(t, lo.ToPtr[uint32](1<<8|1<<2), buildOutput(t, s).Vxlan.Vni2) +} diff --git a/cpdaemon/pkg/cpm/worker_task_builder_validate_test.go b/cpdaemon/pkg/cpm/worker_task_builder_validate_test.go index 2a4fa8da..89ad056b 100644 --- a/cpdaemon/pkg/cpm/worker_task_builder_validate_test.go +++ b/cpdaemon/pkg/cpm/worker_task_builder_validate_test.go @@ -83,6 +83,9 @@ func Test_workerTaskBuilder_rejectsInvalidValues(t *testing.T) { {"vxlan_address_ipv6", PacketChannelType_VXLAN, func(s *StrategyEntry) { s.Address = "2001:db8::1" }, "address"}, {"vxlan_address_v4_mapped", PacketChannelType_VXLAN, func(s *StrategyEntry) { s.Address = "::ffff:10.0.0.1" }, "address"}, {"zmq_address_empty", PacketChannelType_ZMQ, func(s *StrategyEntry) { s.Address = "" }, "address"}, + // An interface task may skip the output-host filter only when its output leaves through + // another interface. + {"nofilter_without_bind_device", PacketChannelType_GRE, func(s *StrategyEntry) { s.Startup = lo.ToPtr("--nofilter") }, "nofilter"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -114,6 +117,10 @@ func Test_workerTaskBuilder_acceptsBoundaryValues(t *testing.T) { }}, // A service tag that is not used (HasServiceTag false) is not checked. {"gre_unused_service_tag", PacketChannelType_GRE, func(s *StrategyEntry) { s.ServiceTag = lo.ToPtr[int32](-1) }}, + {"vxlan_split_bytes_0", PacketChannelType_VXLAN, func(s *StrategyEntry) { + s.HasPacketSplit = true + s.PacketSplitBytes = lo.ToPtr[int32](0) + }}, {"vxlan_split_bytes_65535", PacketChannelType_VXLAN, func(s *StrategyEntry) { s.HasPacketSplit = true s.PacketSplitBytes = lo.ToPtr[int32](65535) diff --git a/cpgolib/cpworker/unix_client_test.go b/cpgolib/cpworker/unix_client_test.go index 4d7cb839..08da3090 100644 --- a/cpgolib/cpworker/unix_client_test.go +++ b/cpgolib/cpworker/unix_client_test.go @@ -34,8 +34,10 @@ type fakeServer struct { t *testing.T socket string handlers map[string]func(args map[string]any) (map[string]any, error) - wg sync.WaitGroup - listener net.Listener + // handshakeReply replaces the {"status":"OK"} handshake reply when set. + handshakeReply string + wg sync.WaitGroup + listener net.Listener } func newFakeServer(t *testing.T) *fakeServer { @@ -97,6 +99,9 @@ func (s *fakeServer) serve(conn net.Conn) { return } resp, _ := json.Marshal(map[string]string{"status": "OK"}) + if s.handshakeReply != "" { + resp = []byte(s.handshakeReply) + } resp = append(resp, '\n') if _, err := conn.Write(resp); err != nil { return @@ -332,3 +337,44 @@ func TestUnixClient_DefaultTimeoutApplies(t *testing.T) { t.Errorf("defaultTimeout did not apply: elapsed=%v", elapsed) } } + +// A client must not use a connection whose handshake cpworker did not accept. +func TestUnixClient_HandshakeRejected(t *testing.T) { + for name, reply := range map[string]string{ + "error_status": `{"status":"ERROR","message":"unsupported version"}`, + "invalid_json": `not json`, + } { + t.Run(name, func(t *testing.T) { + srv := newFakeServer(t) + srv.handshakeReply = reply + defer srv.Close() + + client, err := NewClient(srv.connStr()) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + if err := client.Dial(context.Background()); err == nil { + t.Fatal("expected Dial to fail on a rejected handshake") + } + }) + } +} + +// A timeout of 0 falls back to DefaultTimeout instead of failing every request at once. +func TestUnixClient_ZeroTimeoutUsesDefault(t *testing.T) { + srv := newFakeServer(t) + defer srv.Close() + srv.handlers["ping"] = func(args map[string]any) (map[string]any, error) { return nil, nil } + + client, err := NewClientWithTimeout(srv.connStr(), 0) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + if _, err := client.Ping(context.Background()); err != nil { + t.Fatalf("ping: %v", err) + } +}