test(cpdaemon,cpgolib): check cpm config mapping, buffer sizing, sync state and handshake

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)
+		})
+	}
+}

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())
+}

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,13 +122,15 @@ 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 {
 	mu sync.Mutex
 
 	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)
+}

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)
+}

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")
+}