From d63bce71186f68f70dc0a1c77f911fa5bcac6d87 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sat, 2 May 2026 09:45:04 -0400 Subject: [PATCH] feat(asap-precompute-go): HttpPollChannel + OpAmpChannel (Phase 2 step 2.10) - HttpPollChannel: ETag-aware HTTP polling with bearer-token rotation, configurable Ack endpoint, race-tested concurrency. - OpAmpChannel: interface-satisfying stub; full OpAMP integration deferred to Phase 5 per ADR-0003. - httptest-driven coverage for ETag/304, errors, token rotation, cancellation. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../controlchannel/http_poll.go | 348 +++++++++++++ .../controlchannel/http_poll_test.go | 478 ++++++++++++++++++ asap-precompute-go/controlchannel/opamp.go | 99 ++++ .../controlchannel/opamp_test.go | 69 +++ 4 files changed, 994 insertions(+) create mode 100644 asap-precompute-go/controlchannel/http_poll.go create mode 100644 asap-precompute-go/controlchannel/http_poll_test.go create mode 100644 asap-precompute-go/controlchannel/opamp.go create mode 100644 asap-precompute-go/controlchannel/opamp_test.go diff --git a/asap-precompute-go/controlchannel/http_poll.go b/asap-precompute-go/controlchannel/http_poll.go new file mode 100644 index 00000000..ff1ce5d7 --- /dev/null +++ b/asap-precompute-go/controlchannel/http_poll.go @@ -0,0 +1,348 @@ +package controlchannel + +// HttpPollChannel is the controller-driven, pull-based ControlChannel +// implementation. The runtime polls a configurable HTTP endpoint at a +// fixed interval; the server returns a JSON-encoded PrecomputeConfigSet +// when the plan has changed and HTTP 304 (Not Modified) when it has +// not. Plan-version acknowledgement is forwarded to a separate (and +// optional) Ack endpoint via POST. +// +// # Wire format +// +// The plan body is the standard encoding/json marshalling of +// precompute.PrecomputeConfigSet. The struct has no explicit json tags +// today, so field names follow Go-export casing (Version, Configs, +// AggID, …); a future ADR may pin a more controller-stable schema, but +// the round-trip via encoding/json is the contract for now. +// +// Plan freshness is tracked via the HTTP ETag header. The first +// successful 200 response stores the ETag; subsequent polls send it +// back as If-None-Match and treat 304 as "no change". A 200 with an +// ETag that differs from the cached value (or no ETag at all) returns +// the freshly-decoded set. +// +// # Ack +// +// Ack(planVersion) POSTs a tiny JSON body of the shape +// `{"plan_version": N}` to cfg.AckURL. When AckURL is empty, Ack is a +// no-op. Ack errors are logged via cfg.Logger but do not surface +// through the (errorless) ControlChannel interface; callers can +// inspect HttpPollChannel.LastErr() for the most recent error. +// +// # Bearer token rotation +// +// cfg.BearerTokenFile, when set, is read on every poll and ack so that +// rotated credentials are picked up without restart. A read failure is +// treated as a poll error and the request is skipped that tick. +// +// # Concurrency +// +// A single HttpPollChannel is safe to share across goroutines: Poll, +// Ack, Close, and LastErr are all guarded by an internal mutex on +// the etag / closed / lastErr fields. + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "os" + "strings" + "sync" + "time" + + precompute "github.com/ProjectASAP/asap-precompute-go" +) + +// ErrChannelClosed is returned (via LastErr) after Close has been +// called and another Poll/Ack is attempted. +var ErrChannelClosed = errors.New("controlchannel: channel closed") + +// HttpPollConfig configures HttpPollChannel. +type HttpPollConfig struct { + // URL is the GET endpoint that returns the JSON-encoded + // PrecomputeConfigSet. Required. + URL string + + // AckURL is the POST endpoint that receives plan-version + // acknowledgements. Optional — when empty, Ack is a no-op. + AckURL string + + // Interval is the suggested poll cadence. HttpPollChannel + // itself does not run a goroutine; callers are responsible for + // invoking Poll on a ticker. Interval is exposed here so that + // platform adapters can reach for one canonical value. + Interval time.Duration + + // Timeout is the per-request HTTP timeout. Used to construct + // the default *http.Client when Client is nil. + Timeout time.Duration + + // Headers are extra static headers attached to every Poll and + // Ack request (e.g. tenant identifier, user-agent override). + Headers map[string]string + + // BearerTokenFile, when set, is read on every request; the + // trimmed contents become the value of an + // `Authorization: Bearer ` header. Re-reading per + // request is intentional — credential rotation should not + // require a restart. + BearerTokenFile string + + // Client overrides the default *http.Client. When nil, a + // client with Timeout is constructed. + Client *http.Client + + // Logger receives one-line log messages for non-fatal errors. + // Defaults to log.Default(). + Logger *log.Logger +} + +// HttpPollChannel implements ControlChannel by polling an HTTP +// endpoint. See package doc for wire-format details. +type HttpPollChannel struct { + cfg HttpPollConfig + client *http.Client + logger *log.Logger + + mu sync.Mutex + etag string + closed bool + lastErr error + closeCtx context.Context + cancel context.CancelFunc +} + +// NewHttpPollChannel constructs a new HttpPollChannel. cfg.URL is +// required; everything else has reasonable defaults. +func NewHttpPollChannel(cfg HttpPollConfig) (*HttpPollChannel, error) { + if cfg.URL == "" { + return nil, errors.New("controlchannel: HttpPollConfig.URL is required") + } + if cfg.Timeout <= 0 { + cfg.Timeout = 10 * time.Second + } + if cfg.Interval <= 0 { + cfg.Interval = 30 * time.Second + } + client := cfg.Client + if client == nil { + client = &http.Client{Timeout: cfg.Timeout} + } + logger := cfg.Logger + if logger == nil { + logger = log.Default() + } + ctx, cancel := context.WithCancel(context.Background()) + return &HttpPollChannel{ + cfg: cfg, + client: client, + logger: logger, + closeCtx: ctx, + cancel: cancel, + }, nil +} + +// Poll issues a GET to cfg.URL and returns the freshly-decoded +// PrecomputeConfigSet when the server reports a change. Returns nil +// for "no change" (HTTP 304), for errors (the error is stashed in +// LastErr and logged), and after Close. +// +// The ControlChannel interface is errorless; non-fatal errors do not +// stop polling. Callers that want stronger guarantees can read +// LastErr after each call. +func (h *HttpPollChannel) Poll() *precompute.PrecomputeConfigSet { + h.mu.Lock() + if h.closed { + h.lastErr = ErrChannelClosed + ctx := h.closeCtx + h.mu.Unlock() + _ = ctx + return nil + } + etag := h.etag + ctx := h.closeCtx + h.mu.Unlock() + + reqCtx, cancel := context.WithTimeout(ctx, h.cfg.Timeout) + defer cancel() + + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, h.cfg.URL, nil) + if err != nil { + h.recordErr(fmt.Errorf("controlchannel: build request: %w", err)) + return nil + } + req.Header.Set("Accept", "application/json") + if etag != "" { + req.Header.Set("If-None-Match", etag) + } + if err := h.applyAuth(req); err != nil { + h.recordErr(err) + return nil + } + for k, v := range h.cfg.Headers { + req.Header.Set(k, v) + } + + resp, err := h.client.Do(req) + if err != nil { + h.recordErr(fmt.Errorf("controlchannel: GET %s: %w", h.cfg.URL, err)) + return nil + } + defer resp.Body.Close() + + switch { + case resp.StatusCode == http.StatusNotModified: + h.clearErr() + return nil + case resp.StatusCode >= 200 && resp.StatusCode < 300: + body, err := io.ReadAll(resp.Body) + if err != nil { + h.recordErr(fmt.Errorf("controlchannel: read body: %w", err)) + return nil + } + var set precompute.PrecomputeConfigSet + if err := json.Unmarshal(body, &set); err != nil { + h.recordErr(fmt.Errorf("controlchannel: decode body: %w", err)) + return nil + } + newEtag := resp.Header.Get("ETag") + h.mu.Lock() + h.etag = newEtag + h.lastErr = nil + h.mu.Unlock() + return &set + default: + // Drain a small slice for diagnostics, then error out. + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 256)) + h.recordErr(fmt.Errorf("controlchannel: GET %s: status %d: %s", + h.cfg.URL, resp.StatusCode, strings.TrimSpace(string(snippet)))) + return nil + } +} + +// Ack POSTs the plan version to cfg.AckURL. When AckURL is empty Ack +// is a no-op. Errors are logged and recorded on LastErr; the (errorless) +// ControlChannel interface does not surface them directly. +func (h *HttpPollChannel) Ack(planVersion uint64) { + h.mu.Lock() + if h.closed { + h.lastErr = ErrChannelClosed + h.mu.Unlock() + return + } + ctx := h.closeCtx + h.mu.Unlock() + + if h.cfg.AckURL == "" { + return + } + + reqCtx, cancel := context.WithTimeout(ctx, h.cfg.Timeout) + defer cancel() + + body, err := json.Marshal(struct { + PlanVersion uint64 `json:"plan_version"` + }{PlanVersion: planVersion}) + if err != nil { + // Should be impossible for a uint64 wrapper, but be safe. + h.recordErr(fmt.Errorf("controlchannel: marshal ack: %w", err)) + return + } + + req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, h.cfg.AckURL, bytes.NewReader(body)) + if err != nil { + h.recordErr(fmt.Errorf("controlchannel: build ack: %w", err)) + return + } + req.Header.Set("Content-Type", "application/json") + if err := h.applyAuth(req); err != nil { + h.recordErr(err) + return + } + for k, v := range h.cfg.Headers { + req.Header.Set(k, v) + } + + resp, err := h.client.Do(req) + if err != nil { + h.recordErr(fmt.Errorf("controlchannel: POST %s: %w", h.cfg.AckURL, err)) + return + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 256)) + h.recordErr(fmt.Errorf("controlchannel: POST %s: status %d: %s", + h.cfg.AckURL, resp.StatusCode, strings.TrimSpace(string(snippet)))) + return + } + h.clearErr() +} + +// Close stops accepting new Poll/Ack calls and cancels any in-flight +// request context. Subsequent Poll/Ack calls record ErrChannelClosed +// on LastErr and return without I/O. Close is idempotent. +func (h *HttpPollChannel) Close() error { + h.mu.Lock() + if h.closed { + h.mu.Unlock() + return nil + } + h.closed = true + cancel := h.cancel + h.mu.Unlock() + if cancel != nil { + cancel() + } + return nil +} + +// LastErr returns the most recent non-nil error observed by Poll/Ack. +// Reading LastErr does not clear it; a subsequent successful call +// resets it to nil. +func (h *HttpPollChannel) LastErr() error { + h.mu.Lock() + defer h.mu.Unlock() + return h.lastErr +} + +func (h *HttpPollChannel) recordErr(err error) { + if err == nil { + return + } + h.mu.Lock() + h.lastErr = err + h.mu.Unlock() + h.logger.Printf("%v", err) +} + +func (h *HttpPollChannel) clearErr() { + h.mu.Lock() + h.lastErr = nil + h.mu.Unlock() +} + +func (h *HttpPollChannel) applyAuth(req *http.Request) error { + if h.cfg.BearerTokenFile == "" { + return nil + } + raw, err := os.ReadFile(h.cfg.BearerTokenFile) + if err != nil { + return fmt.Errorf("controlchannel: read bearer token file %q: %w", + h.cfg.BearerTokenFile, err) + } + token := strings.TrimSpace(string(raw)) + if token == "" { + return fmt.Errorf("controlchannel: bearer token file %q is empty", + h.cfg.BearerTokenFile) + } + req.Header.Set("Authorization", "Bearer "+token) + return nil +} + +// Compile-time check. +var _ ControlChannel = (*HttpPollChannel)(nil) diff --git a/asap-precompute-go/controlchannel/http_poll_test.go b/asap-precompute-go/controlchannel/http_poll_test.go new file mode 100644 index 00000000..0864c0ee --- /dev/null +++ b/asap-precompute-go/controlchannel/http_poll_test.go @@ -0,0 +1,478 @@ +package controlchannel + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + precompute "github.com/ProjectASAP/asap-precompute-go" +) + +// makePlanBody returns the JSON-encoded form of a PrecomputeConfigSet +// with the given version and a single config — enough to exercise the +// happy-path round-trip. +func makePlanBody(t *testing.T, version uint64) []byte { + t.Helper() + set := precompute.PrecomputeConfigSet{ + Version: version, + Configs: []precompute.PrecomputeConfig{{ + AggID: precompute.AggId(version * 100), + }}, + } + body, err := json.Marshal(set) + if err != nil { + t.Fatalf("marshal plan: %v", err) + } + return body +} + +// TestHttpPoll_ETag304Roundtrip covers the happy path: first Poll +// returns a fresh plan with a stored ETag; second Poll with the same +// ETag receives a 304 and returns nil; a third Poll after the server +// rotates the ETag returns the new plan. +func TestHttpPoll_ETag304Roundtrip(t *testing.T) { + t.Parallel() + + const etagV1 = `"v1"` + const etagV2 = `"v2"` + var serverVersion atomic.Uint64 + serverVersion.Store(1) + var requestCount atomic.Int32 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestCount.Add(1) + v := serverVersion.Load() + var currentEtag string + switch v { + case 1: + currentEtag = etagV1 + default: + currentEtag = etagV2 + } + if r.Header.Get("If-None-Match") == currentEtag { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("ETag", currentEtag) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(makePlanBody(t, v)) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL, + Timeout: 2 * time.Second, + Interval: 100 * time.Millisecond, + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + // First poll: fresh plan v1. + got := ch.Poll() + if got == nil { + t.Fatalf("first poll: want plan, got nil (lastErr=%v)", ch.LastErr()) + } + if got.Version != 1 { + t.Errorf("first poll version: want 1, got %d", got.Version) + } + if ch.LastErr() != nil { + t.Errorf("first poll lastErr: want nil, got %v", ch.LastErr()) + } + + // Second poll with same server version: 304 → nil, no error. + got = ch.Poll() + if got != nil { + t.Errorf("second poll: want nil (304), got %+v", got) + } + if ch.LastErr() != nil { + t.Errorf("304 poll lastErr: want nil, got %v", ch.LastErr()) + } + + // Server rotates: third poll picks up plan v2. + serverVersion.Store(2) + got = ch.Poll() + if got == nil { + t.Fatalf("third poll: want plan v2, got nil (lastErr=%v)", ch.LastErr()) + } + if got.Version != 2 { + t.Errorf("third poll version: want 2, got %d", got.Version) + } + + if requestCount.Load() != 3 { + t.Errorf("server requests: want 3, got %d", requestCount.Load()) + } +} + +// TestHttpPoll_ServerError exercises the 500-response path: Poll +// returns nil and stashes the error on LastErr. +func TestHttpPoll_ServerError(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL, + Timeout: time.Second, + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + if got := ch.Poll(); got != nil { + t.Fatalf("poll: want nil on 5xx, got %+v", got) + } + if ch.LastErr() == nil { + t.Fatalf("LastErr: want non-nil on 5xx, got nil") + } + if !strings.Contains(ch.LastErr().Error(), "500") { + t.Errorf("LastErr should mention status 500: %v", ch.LastErr()) + } +} + +// TestHttpPoll_DecodeError covers malformed JSON: 200 OK but the body +// can't be parsed. +func TestHttpPoll_DecodeError(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("ETag", `"x"`) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("not json")) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{URL: srv.URL, Timeout: time.Second}) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + if got := ch.Poll(); got != nil { + t.Fatalf("poll: want nil on decode error, got %+v", got) + } + if ch.LastErr() == nil || !strings.Contains(ch.LastErr().Error(), "decode") { + t.Fatalf("LastErr: want decode error, got %v", ch.LastErr()) + } +} + +// TestHttpPoll_BearerTokenRotation writes a token, polls, rewrites the +// token, polls again — the test handler asserts both tokens were +// observed (i.e. the file is re-read per request). +func TestHttpPoll_BearerTokenRotation(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + tokFile := filepath.Join(dir, "token") + if err := os.WriteFile(tokFile, []byte("token-v1"), 0o600); err != nil { + t.Fatalf("write token v1: %v", err) + } + + var ( + mu sync.Mutex + seen = map[string]bool{} + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + auth := r.Header.Get("Authorization") + mu.Lock() + seen[auth] = true + mu.Unlock() + w.Header().Set("ETag", auth) // rotate ETag with token so 304 doesn't kick in + w.WriteHeader(http.StatusOK) + _, _ = w.Write(makePlanBody(t, 1)) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL, + Timeout: 2 * time.Second, + BearerTokenFile: tokFile, + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + if got := ch.Poll(); got == nil { + t.Fatalf("poll v1: want plan, got nil (lastErr=%v)", ch.LastErr()) + } + if err := os.WriteFile(tokFile, []byte("token-v2"), 0o600); err != nil { + t.Fatalf("write token v2: %v", err) + } + if got := ch.Poll(); got == nil { + t.Fatalf("poll v2: want plan, got nil (lastErr=%v)", ch.LastErr()) + } + + mu.Lock() + defer mu.Unlock() + if !seen["Bearer token-v1"] { + t.Errorf("server never saw token-v1; seen=%v", seen) + } + if !seen["Bearer token-v2"] { + t.Errorf("server never saw token-v2; seen=%v", seen) + } +} + +// TestHttpPoll_BearerTokenFileMissing covers the failure path: the +// token file vanishes between polls. Poll should return nil and stash +// the error on LastErr. +func TestHttpPoll_BearerTokenFileMissing(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Errorf("server should not be reached when token file is missing") + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL, + Timeout: time.Second, + BearerTokenFile: filepath.Join(t.TempDir(), "does-not-exist"), + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + if got := ch.Poll(); got != nil { + t.Fatalf("poll: want nil on missing token, got %+v", got) + } + if ch.LastErr() == nil || !strings.Contains(ch.LastErr().Error(), "bearer token") { + t.Fatalf("LastErr: want bearer-token error, got %v", ch.LastErr()) + } +} + +// TestHttpPoll_ContextCancellation wires a slow server and calls Close +// from another goroutine while a Poll is in flight. The Poll must +// return nil promptly and surface a context-cancellation error. +func TestHttpPoll_ContextCancellation(t *testing.T) { + t.Parallel() + + released := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-r.Context().Done(): + case <-released: + case <-time.After(5 * time.Second): + } + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + defer close(released) + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL, + Timeout: 5 * time.Second, + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + + done := make(chan struct{}) + var got *precompute.PrecomputeConfigSet + go func() { + defer close(done) + got = ch.Poll() + }() + + // Give the request a moment to dispatch, then close. + time.Sleep(50 * time.Millisecond) + if err := ch.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatalf("Poll did not return after Close") + } + if got != nil { + t.Errorf("Poll: want nil on cancel, got %+v", got) + } + if ch.LastErr() == nil || + !(strings.Contains(ch.LastErr().Error(), "context canceled") || + errors.Is(ch.LastErr(), context.Canceled)) { + t.Errorf("LastErr: want context-canceled flavor, got %v", ch.LastErr()) + } +} + +// TestHttpPoll_AckURL exercises the Ack path: a non-empty AckURL gets +// a POST with the plan_version JSON body; an empty AckURL is a no-op. +func TestHttpPoll_AckURL(t *testing.T) { + t.Parallel() + + type ackBody struct { + PlanVersion uint64 `json:"plan_version"` + } + gotBody := make(chan ackBody, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("ack: want POST, got %s", r.Method) + } + raw, _ := io.ReadAll(r.Body) + var b ackBody + if err := json.Unmarshal(raw, &b); err != nil { + t.Errorf("ack: decode body: %v", err) + } + gotBody <- b + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL + "/poll", + AckURL: srv.URL + "/ack", + Timeout: time.Second, + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + ch.Ack(42) + select { + case b := <-gotBody: + if b.PlanVersion != 42 { + t.Errorf("ack body: want 42, got %d", b.PlanVersion) + } + case <-time.After(time.Second): + t.Fatalf("ack: server never received POST") + } + if ch.LastErr() != nil { + t.Errorf("ack lastErr: want nil, got %v", ch.LastErr()) + } + + // AckURL == "" → no-op (no panic, no error, no request). + chNoAck, err := NewHttpPollChannel(HttpPollConfig{URL: srv.URL, Timeout: time.Second}) + if err != nil { + t.Fatalf("NewHttpPollChannel no-ack: %v", err) + } + defer chNoAck.Close() + chNoAck.Ack(7) // must not panic + if chNoAck.LastErr() != nil { + t.Errorf("no-ack lastErr: want nil, got %v", chNoAck.LastErr()) + } +} + +// TestHttpPoll_AckErrorStatus exercises the 5xx-on-ack path: Ack +// records the error on LastErr but does not panic. +func TestHttpPoll_AckErrorStatus(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "no", http.StatusBadGateway) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{ + URL: srv.URL, + AckURL: srv.URL, + Timeout: time.Second, + }) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + defer ch.Close() + + ch.Ack(99) + if ch.LastErr() == nil || !strings.Contains(ch.LastErr().Error(), "502") { + t.Errorf("LastErr: want 502 error, got %v", ch.LastErr()) + } +} + +// TestHttpPoll_AfterClose verifies that Poll/Ack after Close are +// inert and report ErrChannelClosed via LastErr. +func TestHttpPoll_AfterClose(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Errorf("server should not be reached after Close") + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{URL: srv.URL, Timeout: time.Second}) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + if err := ch.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if got := ch.Poll(); got != nil { + t.Errorf("Poll after Close: want nil, got %+v", got) + } + if !errors.Is(ch.LastErr(), ErrChannelClosed) { + t.Errorf("LastErr after Close: want ErrChannelClosed, got %v", ch.LastErr()) + } + ch.Ack(1) + if !errors.Is(ch.LastErr(), ErrChannelClosed) { + t.Errorf("LastErr after Ack post-Close: want ErrChannelClosed, got %v", ch.LastErr()) + } + // Close is idempotent. + if err := ch.Close(); err != nil { + t.Errorf("second Close: want nil, got %v", err) + } +} + +// TestHttpPoll_ConstructorValidation ensures a missing URL is rejected +// at construction time. +func TestHttpPoll_ConstructorValidation(t *testing.T) { + t.Parallel() + + if _, err := NewHttpPollChannel(HttpPollConfig{}); err == nil { + t.Fatalf("NewHttpPollChannel(empty): want error, got nil") + } +} + +// TestHttpPoll_ConcurrentPollClose exercises the race detector: many +// concurrent Polls overlap with a Close. Nothing should crash, and +// after Close every Poll should return nil. +func TestHttpPoll_ConcurrentPollClose(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("ETag", `"x"`) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(makePlanBody(t, 1)) + })) + defer srv.Close() + + ch, err := NewHttpPollChannel(HttpPollConfig{URL: srv.URL, Timeout: 2 * time.Second}) + if err != nil { + t.Fatalf("NewHttpPollChannel: %v", err) + } + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 5; j++ { + _ = ch.Poll() + ch.Ack(uint64(j)) + } + }() + } + // Race the Close against the goroutines. + go func() { + time.Sleep(20 * time.Millisecond) + _ = ch.Close() + }() + wg.Wait() +} diff --git a/asap-precompute-go/controlchannel/opamp.go b/asap-precompute-go/controlchannel/opamp.go new file mode 100644 index 00000000..fd691a12 --- /dev/null +++ b/asap-precompute-go/controlchannel/opamp.go @@ -0,0 +1,99 @@ +package controlchannel + +import ( + "errors" + "log" + "sync" + + precompute "github.com/ProjectASAP/asap-precompute-go" +) + +// OpAmpChannel is the OpAMP-backed ControlChannel implementation. +// +// TODO(phase-5): wire real OpAMP client (github.com/open-telemetry/ +// opamp-go/client) per docs/adr/adr-0003-adapter-trait-and-control-channel.md; +// today this is an interface-satisfying stub so the controller side +// can compile against the trait without forcing a transitive +// opamp-go dep into Phase 2. +type OpAmpChannel struct { + cfg OpAmpConfig + logger *log.Logger + + mu sync.Mutex + closed bool + warnedOnce bool +} + +// OpAmpConfig configures OpAmpChannel. Real fields will grow as the +// Phase 5 wiring lands. +type OpAmpConfig struct { + // ServerEndpoint is the OpAMP supervisor's URL (ws/wss/http(s)). + // Required. + ServerEndpoint string + + // InstanceUid is the agent's stable identifier used by the + // supervisor to address this runtime. + InstanceUid string + + // Capabilities is the bitmask of OpAMP capabilities advertised + // to the supervisor (see opamp-go's protobuf for values). + Capabilities uint64 + + // Logger receives one-line log messages. Defaults to log.Default(). + Logger *log.Logger +} + +// NewOpAmpChannel constructs a new OpAmpChannel. Validates that +// ServerEndpoint is non-empty; the rest of the fields are accepted as-is +// pending real OpAMP integration. +func NewOpAmpChannel(cfg OpAmpConfig) (*OpAmpChannel, error) { + if cfg.ServerEndpoint == "" { + return nil, errors.New("controlchannel: OpAmpConfig.ServerEndpoint is required") + } + logger := cfg.Logger + if logger == nil { + logger = log.Default() + } + return &OpAmpChannel{cfg: cfg, logger: logger}, nil +} + +// Poll satisfies ControlChannel; the stub never reports a change. A +// single warning is emitted per instance to avoid log spam. +func (o *OpAmpChannel) Poll() *precompute.PrecomputeConfigSet { + o.mu.Lock() + warn := !o.warnedOnce && !o.closed + o.warnedOnce = true + o.mu.Unlock() + if warn { + o.logger.Printf("OpAMP control channel: not yet wired, returning no-op (endpoint=%s)", + o.cfg.ServerEndpoint) + } + return nil +} + +// Ack is a no-op for the stub. Logged at debug-level via the configured +// logger; the standard library log.Logger has no levels, so we simply +// write a one-line debug-style message. +func (o *OpAmpChannel) Ack(planVersion uint64) { + o.mu.Lock() + closed := o.closed + o.mu.Unlock() + if closed { + return + } + // Intentionally low-volume — Ack is rare relative to Poll. + o.logger.Printf("debug: OpAMP control channel: stub Ack(plan_version=%d) (endpoint=%s)", + planVersion, o.cfg.ServerEndpoint) +} + +// Close is a no-op for the stub but flips the internal flag so Poll/Ack +// can short-circuit after shutdown. Idempotent. +func (o *OpAmpChannel) Close() error { + o.mu.Lock() + o.closed = true + o.mu.Unlock() + return nil +} + +// Compile-time check. +var _ ControlChannel = (*OpAmpChannel)(nil) diff --git a/asap-precompute-go/controlchannel/opamp_test.go b/asap-precompute-go/controlchannel/opamp_test.go new file mode 100644 index 00000000..15382100 --- /dev/null +++ b/asap-precompute-go/controlchannel/opamp_test.go @@ -0,0 +1,69 @@ +package controlchannel + +import ( + "testing" +) + +func TestOpAmp_ConstructorValidatesEndpoint(t *testing.T) { + t.Parallel() + if _, err := NewOpAmpChannel(OpAmpConfig{}); err == nil { + t.Fatalf("NewOpAmpChannel(empty): want error, got nil") + } + ch, err := NewOpAmpChannel(OpAmpConfig{ServerEndpoint: "http://localhost:4320"}) + if err != nil { + t.Fatalf("NewOpAmpChannel: %v", err) + } + if ch == nil { + t.Fatalf("NewOpAmpChannel: want non-nil channel") + } +} + +func TestOpAmp_PollReturnsNoOp(t *testing.T) { + t.Parallel() + ch, err := NewOpAmpChannel(OpAmpConfig{ServerEndpoint: "http://localhost:4320"}) + if err != nil { + t.Fatalf("NewOpAmpChannel: %v", err) + } + if got := ch.Poll(); got != nil { + t.Errorf("Poll: want nil, got %+v", got) + } + // Second Poll should still return nil and (silently) skip the + // once-per-instance warn log. + if got := ch.Poll(); got != nil { + t.Errorf("second Poll: want nil, got %+v", got) + } +} + +func TestOpAmp_AckIsNoOp(t *testing.T) { + t.Parallel() + ch, err := NewOpAmpChannel(OpAmpConfig{ServerEndpoint: "http://localhost:4320"}) + if err != nil { + t.Fatalf("NewOpAmpChannel: %v", err) + } + // Must not panic and must accept any plan version. + ch.Ack(0) + ch.Ack(42) +} + +func TestOpAmp_CloseIsIdempotent(t *testing.T) { + t.Parallel() + ch, err := NewOpAmpChannel(OpAmpConfig{ServerEndpoint: "http://localhost:4320"}) + if err != nil { + t.Fatalf("NewOpAmpChannel: %v", err) + } + if err := ch.Close(); err != nil { + t.Errorf("first Close: want nil, got %v", err) + } + if err := ch.Close(); err != nil { + t.Errorf("second Close: want nil, got %v", err) + } + // Poll after Close should still return nil, no panic. + if got := ch.Poll(); got != nil { + t.Errorf("Poll after Close: want nil, got %+v", got) + } +} + +func TestOpAmp_SatisfiesControlChannel(t *testing.T) { + t.Parallel() + var _ ControlChannel = (*OpAmpChannel)(nil) +}