From 91eee022439b4ab6a6b4fb4bdfc30bdcb682c921 Mon Sep 17 00:00:00 2001 From: tursom Date: Fri, 26 Jun 2026 12:32:45 +0800 Subject: [PATCH] feat(plugin): add observability operations --- cmd/gateway/admin_api.go | 35 +- cmd/gateway/admin_frontend/src/types.ts | 16 + .../admin_frontend/src/views/plugins.ts | 51 +- cmd/gateway/admin_plugin_handlers.go | 130 ++ cmd/gateway/plugin.go | 78 + .../mc-auth-proxy/cmd/render-manifest/main.go | 20 + examples/plugins/mc-auth-proxy/main.go | 44 +- examples/plugins/mc-auth-proxy/main_test.go | 68 + examples/plugins/mc-auth-proxy/manifest.json | 33 +- internal/admindb/db.go | 80 + internal/adminhttp/api.go | 45 +- internal/adminhttp/api_test.go | 39 +- internal/pluginmanager/artifact.go | 92 +- internal/pluginmanager/manager.go | 127 +- internal/pluginmanager/manager_test.go | 153 ++ internal/pluginmanager/operations.go | 1590 +++++++++++++++++ internal/pluginmanager/repository.go | 333 ++++ internal/pluginmanager/types.go | 281 ++- plugin/api/api.go | 83 + plugin/api/api_test.go | 66 + plugin/api/hook.go | 3 + 21 files changed, 3310 insertions(+), 57 deletions(-) create mode 100644 internal/pluginmanager/operations.go diff --git a/cmd/gateway/admin_api.go b/cmd/gateway/admin_api.go index bf78f3d..c93e5d7 100644 --- a/cmd/gateway/admin_api.go +++ b/cmd/gateway/admin_api.go @@ -29,21 +29,24 @@ func newAdminAPIHandler() http.HandlerFunc { AuditLogs: handleAdminAuditLogs, - PluginArtifacts: handleAdminPluginArtifacts, - PluginArtifact: handleAdminPluginArtifact, - PluginSources: handleAdminPluginSources, - PluginBuilds: handleAdminPluginBuilds, - PluginBuild: handleAdminPluginBuild, - PluginGC: handleAdminPluginGC, - PluginsList: handleAdminPluginsList, - PluginItem: handleAdminPluginItem, - PluginAction: handleAdminPluginAction, - PluginConfig: handleAdminPluginConfig, - PluginSecrets: handleAdminPluginSecrets, - PluginRollback: handleAdminPluginRollback, - PluginDraining: handleAdminPluginDraining, - PluginDispatch: handleAdminPluginDispatchPlan, - PluginGovernance: handleAdminPluginGovernance, - PluginAdvisories: handleAdminPluginAdvisories, + PluginArtifacts: handleAdminPluginArtifacts, + PluginArtifact: handleAdminPluginArtifact, + PluginSources: handleAdminPluginSources, + PluginBuilds: handleAdminPluginBuilds, + PluginBuild: handleAdminPluginBuild, + PluginGC: handleAdminPluginGC, + PluginOperationsGC: handleAdminPluginOperationsGC, + PluginsList: handleAdminPluginsList, + PluginItem: handleAdminPluginItem, + PluginAction: handleAdminPluginAction, + PluginConfig: handleAdminPluginConfig, + PluginSecrets: handleAdminPluginSecrets, + PluginRollback: handleAdminPluginRollback, + PluginOperations: handleAdminPluginOperations, + PluginDraining: handleAdminPluginDraining, + PluginDispatch: handleAdminPluginDispatchPlan, + PluginGovernance: handleAdminPluginGovernance, + PluginAdvisories: handleAdminPluginAdvisories, + PluginDiagnostics: handleAdminPluginDiagnostics, }) } diff --git a/cmd/gateway/admin_frontend/src/types.ts b/cmd/gateway/admin_frontend/src/types.ts index 217b996..346d089 100644 --- a/cmd/gateway/admin_frontend/src/types.ts +++ b/cmd/gateway/admin_frontend/src/types.ts @@ -192,6 +192,22 @@ export interface PluginView { updated_at?: number; } +export interface PluginOperations { + handlers?: Record[]; + builds?: Record[]; + events?: Record[]; + custom_metrics?: Record[]; + logs?: Record[]; + traces?: Record[]; + background_tasks?: Record[]; + plugin_data?: Record[]; + plugin_files?: Record[]; + external_dependencies?: Record[]; + gc?: Record[]; + event_queue?: Record; + diagnostics?: Record[]; +} + export interface PluginDryRunResult { ok: boolean; restart_required: boolean; diff --git a/cmd/gateway/admin_frontend/src/views/plugins.ts b/cmd/gateway/admin_frontend/src/views/plugins.ts index 7784325..a49366d 100644 --- a/cmd/gateway/admin_frontend/src/views/plugins.ts +++ b/cmd/gateway/admin_frontend/src/views/plugins.ts @@ -3,7 +3,7 @@ import { showAlert } from "../alerts.js"; import { badge, el, escapeAttr, escapeHTML, getFormInput } from "../dom.js"; import { isAdmin } from "../session.js"; import { state } from "../state.js"; -import type { PluginArtifact, PluginBuild, PluginDryRunResult, PluginProxyConnection, PluginSecret, PluginSnapshot, PluginView } from "../types.js"; +import type { PluginArtifact, PluginBuild, PluginDryRunResult, PluginOperations, PluginProxyConnection, PluginSecret, PluginSnapshot, PluginView } from "../types.js"; interface PluginsResponse { plugins?: PluginView[]; @@ -33,6 +33,13 @@ interface SnapshotDiffResponse { }; } +interface OperationsResponse { + operations?: PluginOperations; + candidates?: Record[]; + diagnostic?: Record; + summary?: Record; +} + export async function loadPlugins(): Promise { try { const [data, artifacts, builds] = await Promise.all([ @@ -209,6 +216,15 @@ export function renderPluginDetail(plugin: PluginView | null = selectedPlugin())

Dispatch plan

${escapeHTML(formatJSON(plugin.dispatch_summary || []))}
+
+

Operations

+
+ + + +
+

+      

Active proxy connections

${proxyConnectionList(plugin.proxy_connections || [])} @@ -272,6 +288,9 @@ function bindPluginDetailEvents(plugin: PluginView): void { document.getElementById("pluginGovernanceSelfTestBtn")?.addEventListener("click", () => runGovernanceSelfTest(plugin)); document.getElementById("pluginGovernanceBenchmarkBtn")?.addEventListener("click", () => recordGovernanceBenchmark(plugin)); document.getElementById("pluginGovernanceAdvisoryBtn")?.addEventListener("click", () => createArtifactRevokeAdvisory(plugin)); + document.getElementById("pluginOperationsLoadBtn")?.addEventListener("click", () => loadPluginOperations(plugin)); + document.getElementById("pluginOperationsGCDryRunBtn")?.addEventListener("click", () => dryRunOperationsGC(plugin)); + document.getElementById("pluginDiagnosticBtn")?.addEventListener("click", () => loadDiagnosticPackage(plugin)); } async function dryRunConfig(plugin: PluginView): Promise { @@ -534,6 +553,36 @@ async function createArtifactRevokeAdvisory(plugin: PluginView): Promise { } } +async function loadPluginOperations(plugin: PluginView): Promise { + try { + const data = await api(`/plugins/${encodeURIComponent(plugin.id)}/operations`); + el("pluginOperationsOutput").textContent = formatJSON(data.operations || {}); + showAlert(""); + } catch (err) { + showAlert((err as Error).message); + } +} + +async function dryRunOperationsGC(plugin: PluginView): Promise { + try { + const data = await api(`/plugins/${encodeURIComponent(plugin.id)}/operations/gc`); + el("pluginOperationsOutput").textContent = formatJSON(data); + showAlert(""); + } catch (err) { + showAlert((err as Error).message); + } +} + +async function loadDiagnosticPackage(plugin: PluginView): Promise { + try { + const data = await api(`/plugins/${encodeURIComponent(plugin.id)}/operations/diagnostic`); + el("pluginOperationsOutput").textContent = formatJSON(data); + showAlert(""); + } catch (err) { + showAlert((err as Error).message); + } +} + function uploadInventoryDetail(): string { const artifact = selectedArtifact(); if (!artifact) { diff --git a/cmd/gateway/admin_plugin_handlers.go b/cmd/gateway/admin_plugin_handlers.go index 2a536d5..c1a0d7c 100644 --- a/cmd/gateway/admin_plugin_handlers.go +++ b/cmd/gateway/admin_plugin_handlers.go @@ -574,6 +574,136 @@ func handleAdminPluginRollback(w http.ResponseWriter, r *http.Request, rawSegmen adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"plugin": plugin}) } +func handleAdminPluginOperations(w http.ResponseWriter, r *http.Request, rawSegment string) { + session, ok := requireRole(w, r, adminRoleMember) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + pluginID, action, ok := splitPluginSubresource(w, rawSegment, "operations") + if !ok { + return + } + switch { + case r.Method == http.MethodGet && action == "": + snapshot, err := pluginsManager.OperationsSnapshot(r.Context(), pluginID) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"operations": snapshot}) + case r.Method == http.MethodPost && strings.HasPrefix(action, "tasks/") && strings.HasSuffix(action, "/trigger"): + if session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + parts := strings.Split(action, "/") + if len(parts) != 3 { + adminhttp.WriteAPIError(w, http.StatusBadRequest, "invalid task trigger path") + return + } + taskID, err := adminhttp.PathSegment(parts[1]) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + var req struct { + ConfirmToken string `json:"confirm_token"` + } + if !adminhttp.DecodeJSONRequest(w, r, &req) { + return + } + task, err := pluginsManager.TriggerBackgroundTask(r.Context(), session.Username, pluginID, taskID, req.ConfirmToken) + if err != nil { + recordAudit(r.Context(), session.Username, adminhttp.RequestSourceIP(r), "plugin_background_task_trigger", "plugin", pluginID, false, err.Error()) + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + recordAuditMetadata(r.Context(), session.Username, adminhttp.RequestSourceIP(r), "plugin_background_task_trigger", "plugin", pluginID, true, "background task triggered", map[string]any{"task_id": taskID}) + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"task": task}) + case r.Method == http.MethodGet && action == "diagnostic": + data, summary, err := pluginsManager.DiagnosticPackage(r.Context(), session.Username, pluginID) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + var body any + if err := json.Unmarshal(data, &body); err != nil { + body = map[string]any{"error": "diagnostic package could not be decoded"} + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"diagnostic": body, "summary": summary}) + case (r.Method == http.MethodGet || r.Method == http.MethodPost) && action == "gc": + if r.Method == http.MethodPost && session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + dryRun := r.Method == http.MethodGet + candidates, err := pluginsManager.RunOperationsGC(r.Context(), session.Username, pluginID, dryRun) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"dry_run": dryRun, "candidates": candidates}) + default: + adminhttp.WriteAPIError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func handleAdminPluginOperationsGC(w http.ResponseWriter, r *http.Request) { + session, ok := requireRole(w, r, adminRoleMember) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + if r.Method == http.MethodPost && session.Role != adminRoleAdmin { + adminhttp.WriteAPIError(w, http.StatusForbidden, "forbidden") + return + } + dryRun := r.Method == http.MethodGet + candidates, err := pluginsManager.RunOperationsGC(r.Context(), session.Username, r.URL.Query().Get("plugin_id"), dryRun) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusInternalServerError, err.Error()) + return + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"dry_run": dryRun, "candidates": candidates}) +} + +func handleAdminPluginDiagnostics(w http.ResponseWriter, r *http.Request, rawPluginID string) { + session, ok := requireRole(w, r, adminRoleAdmin) + if !ok { + return + } + if pluginsManager == nil { + adminhttp.WriteAPIError(w, http.StatusServiceUnavailable, "plugin manager is not initialized") + return + } + if r.Method != http.MethodGet { + adminhttp.WriteAPIError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + pluginID, err := adminhttp.PathSegment(rawPluginID) + if err != nil { + adminhttp.WriteAPIError(w, http.StatusBadRequest, err.Error()) + return + } + data, summary, err := pluginsManager.DiagnosticPackage(r.Context(), session.Username, pluginID) + if err != nil { + writePluginManagerError(w, err) + return + } + var body any + if err := json.Unmarshal(data, &body); err != nil { + body = map[string]any{"error": "diagnostic package could not be decoded"} + } + adminhttp.WriteJSON(w, http.StatusOK, map[string]any{"diagnostic": body, "summary": summary}) +} + func handleAdminPluginAction(w http.ResponseWriter, r *http.Request, rawSegment string) { session, ok := requireRole(w, r, adminRoleAdmin) if !ok { diff --git a/cmd/gateway/plugin.go b/cmd/gateway/plugin.go index 603cc30..fd4c8a2 100644 --- a/cmd/gateway/plugin.go +++ b/cmd/gateway/plugin.go @@ -1,9 +1,12 @@ package main import ( + "context" + "errors" "net" "plugin" "sync" + "time" "github.com/rs/zerolog/log" "github.com/tursom/mc-gateway/plugin/api" @@ -112,11 +115,86 @@ func (g *Gateway) ExitWaitGroup() *sync.WaitGroup { return &exitWaitGroup } +func (g *Gateway) EmitEvent(ctx context.Context, name string, fields map[string]string) error { + _ = ctx + _ = name + _ = fields + return nil +} + +func (g *Gateway) ObserveMetric(ctx context.Context, name string, value float64, labels map[string]string) error { + _ = ctx + _ = name + _ = value + _ = labels + return nil +} + +func (g *Gateway) Logger() api.Logger { + return noopPluginLogger{} +} + +func (g *Gateway) DataStore() api.DataStore { + return noopPluginDataStore{} +} + +func (g *Gateway) FileStore() api.FileStore { + return noopPluginFileStore{} +} + +func (g *Gateway) ExternalClient(name string) api.ExternalClient { + _ = name + return noopExternalClient{} +} + +func (g *Gateway) RegisterBackgroundTask(task api.BackgroundTask) error { + _ = task + return nil +} + // TestOp implements api.Gateway. func (g *Gateway) TestOp() { panic("unimplemented") } +type noopPluginLogger struct{} + +func (noopPluginLogger) Debug(context.Context, string, map[string]string) {} +func (noopPluginLogger) Info(context.Context, string, map[string]string) {} +func (noopPluginLogger) Warn(context.Context, string, map[string]string) {} +func (noopPluginLogger) Error(context.Context, string, map[string]string) {} + +type noopPluginDataStore struct{} + +func (noopPluginDataStore) Put(context.Context, api.DataRecord) error { return nil } +func (noopPluginDataStore) Get(context.Context, string) (api.DataRecord, error) { + return api.DataRecord{}, errors.New("plugin data store is unavailable") +} +func (noopPluginDataStore) Delete(context.Context, string) error { return nil } + +type noopPluginFileStore struct{} + +func (noopPluginFileStore) ResourcePath(string) (string, error) { + return "", errors.New("plugin file store is unavailable") +} +func (noopPluginFileStore) Write(context.Context, string, string, []byte, string, time.Duration) error { + return nil +} +func (noopPluginFileStore) Read(context.Context, string, string, int64) ([]byte, error) { + return nil, errors.New("plugin file store is unavailable") +} +func (noopPluginFileStore) Delete(context.Context, string, string) error { return nil } + +type noopExternalClient struct{} + +func (noopExternalClient) DoHTTP(context.Context, api.ExternalRequest) (api.ExternalResponse, error) { + return api.ExternalResponse{}, errors.New("external client is unavailable") +} +func (noopExternalClient) DialTCP(context.Context, string, time.Duration) (net.Conn, error) { + return nil, errors.New("external client is unavailable") +} +func (noopExternalClient) HealthCheck(context.Context) error { return nil } + func Handler1[T1, R any](t1 T1) func(func(T1) R) R { return func(acceptor func(T1) R) R { return acceptor(t1) diff --git a/examples/plugins/mc-auth-proxy/cmd/render-manifest/main.go b/examples/plugins/mc-auth-proxy/cmd/render-manifest/main.go index 8347e14..8697575 100644 --- a/examples/plugins/mc-auth-proxy/cmd/render-manifest/main.go +++ b/examples/plugins/mc-auth-proxy/cmd/render-manifest/main.go @@ -70,6 +70,26 @@ func main() { "handler_timeout_ms": 3000, "initial_write_timeout_ms": 1000, }, + "events": []map[string]any{ + {"name": "auth.success", "fields": []string{"result", "mode"}}, + {"name": "auth.failure", "fields": []string{"result", "mode"}}, + }, + "custom_metrics": []map[string]any{ + {"name": "auth.attempts", "type": "counter", "labels": []string{"result", "mode"}}, + }, + "external_dependencies": []map[string]any{ + {"name": "backend", "endpoint": "tcp://", "purpose": "auth", "required": true, "timeout": "3s", "retry": 0, "fail_policy": "fail_closed", "data_classes": []string{"operational"}}, + }, + "background_tasks": []map[string]any{ + {"id": "profile-cache-gc", "name": "Profile cache GC", "mode": "manual", "manual": true, "timeout": "1s"}, + }, + "data_stores": []map[string]any{ + {"name": "profile-cache", "schema_version": 1, "data_class": "profile_cache", "quota_bytes": 1048576, "retention": "24h", "exportable": false}, + }, + "file_stores": []map[string]any{ + {"namespace": "cache", "data_class": "profile_cache", "quota_bytes": 1048576, "retention": "24h"}, + {"namespace": "diagnostic", "data_class": "diagnostic", "quota_bytes": 1048576, "retention": "24h"}, + }, "config_schema": map[string]any{ "type": "object", "properties": map[string]any{ diff --git a/examples/plugins/mc-auth-proxy/main.go b/examples/plugins/mc-auth-proxy/main.go index 194e315..b6dc707 100644 --- a/examples/plugins/mc-auth-proxy/main.go +++ b/examples/plugins/mc-auth-proxy/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "errors" "fmt" @@ -15,7 +16,8 @@ import ( type PluginImpl struct { api.AbstractPlugin - config Config + config Config + gateway api.Gateway } type Config struct { @@ -52,6 +54,7 @@ func (p *PluginImpl) ReloadConfig(config any) error { } func (p *PluginImpl) Init(gateway api.Gateway) error { + p.gateway = gateway return api.RegisterHookHandler( gateway, api.HookUpstreamConnect, @@ -79,6 +82,7 @@ func (p *PluginImpl) handleConn(req api.UpstreamConnectRequest, conn net.Conn) { } handshake := protocol.ParseHandshake(handshakePacket) if handshake.ServerHost == "" || handshake.NextState != 2 { + p.emitAuthEvent(req.Context, "auth.failure", "bad_handshake") _ = writeLoginDisconnect(conn, "Unsupported Minecraft handshake") return } @@ -90,24 +94,29 @@ func (p *PluginImpl) handleConn(req api.UpstreamConnectRequest, conn net.Conn) { } login, err := parseLoginStart(loginPacket) if err != nil || login.Username == "" { + p.emitAuthEvent(req.Context, "auth.failure", "bad_login_start") _ = writeLoginDisconnect(conn, p.config.DisconnectMessage) return } if !p.config.FixtureAccept { + p.emitAuthEvent(req.Context, "auth.failure", "fixture_reject") _ = writeLoginDisconnect(conn, p.config.DisconnectMessage) return } if p.config.Backend == "" { + p.emitAuthEvent(req.Context, "auth.failure", "backend_missing") _ = writeLoginDisconnect(conn, "Fixture accepted but no backend is configured") return } - backend, err := net.Dial("tcp", p.config.Backend) + backend, err := p.gateway.ExternalClient("backend").DialTCP(req.Context, p.config.Backend, 3*time.Second) if err != nil { + p.emitAuthEvent(req.Context, "auth.failure", "backend_unavailable") _ = writeLoginDisconnect(conn, "Backend unavailable") return } + p.emitAuthEvent(req.Context, "auth.success", "fixture_accept") defer backend.Close() _, _ = backend.Write(handshakePacket) _, _ = backend.Write(loginPacket) @@ -115,6 +124,17 @@ func (p *PluginImpl) handleConn(req api.UpstreamConnectRequest, conn net.Conn) { _ = req } +func (p *PluginImpl) emitAuthEvent(ctx context.Context, name, result string) { + if p.gateway == nil { + return + } + _ = p.gateway.EmitEvent(ctx, name, map[string]string{ + "result": result, + "mode": "fixture", + }) + p.gateway.Logger().Info(ctx, name, map[string]string{"result": result}) +} + func readPacketFromConn(conn net.Conn) ([]byte, error) { length, err := readVarIntFromConn(conn) if err != nil { @@ -242,6 +262,26 @@ var manifestJSON = compactJSON(map[string]any{ "handler_timeout_ms": 3000, "initial_write_timeout_ms": 1000, }, + "events": []map[string]any{ + {"name": "auth.success", "fields": []string{"result", "mode"}}, + {"name": "auth.failure", "fields": []string{"result", "mode"}}, + }, + "custom_metrics": []map[string]any{ + {"name": "auth.attempts", "type": "counter", "labels": []string{"result", "mode"}}, + }, + "external_dependencies": []map[string]any{ + {"name": "backend", "endpoint": "tcp://", "purpose": "auth", "required": true, "timeout": "3s", "retry": 0, "fail_policy": "fail_closed", "data_classes": []string{"operational"}}, + }, + "background_tasks": []map[string]any{ + {"id": "profile-cache-gc", "name": "Profile cache GC", "mode": "manual", "manual": true, "timeout": "1s"}, + }, + "data_stores": []map[string]any{ + {"name": "profile-cache", "schema_version": 1, "data_class": "profile_cache", "quota_bytes": 1048576, "retention": "24h", "exportable": false}, + }, + "file_stores": []map[string]any{ + {"namespace": "cache", "data_class": "profile_cache", "quota_bytes": 1048576, "retention": "24h"}, + {"namespace": "diagnostic", "data_class": "diagnostic", "quota_bytes": 1048576, "retention": "24h"}, + }, }) func compactJSON(value any) string { diff --git a/examples/plugins/mc-auth-proxy/main_test.go b/examples/plugins/mc-auth-proxy/main_test.go index ade66d2..cb9bcc2 100644 --- a/examples/plugins/mc-auth-proxy/main_test.go +++ b/examples/plugins/mc-auth-proxy/main_test.go @@ -2,8 +2,11 @@ package main import ( "bytes" + "context" "net" + "sync" "testing" + "time" "github.com/tursom/mc-gateway/plugin/api" ) @@ -13,6 +16,8 @@ func TestFixtureRejectsLoginStart(t *testing.T) { if err := plugin.ReloadConfig(&Config{DisconnectMessage: "fixture rejected"}); err != nil { t.Fatalf("ReloadConfig() error = %v", err) } + gateway := &recordingGateway{} + plugin.gateway = gateway client, server := net.Pipe() done := make(chan struct{}) go func() { @@ -33,6 +38,9 @@ func TestFixtureRejectsLoginStart(t *testing.T) { if !bytes.Contains(response, []byte("fixture rejected")) { t.Fatalf("response = %q, want fixture message", response) } + if len(gateway.events) != 1 || gateway.events[0].name != "auth.failure" { + t.Fatalf("events = %+v, want auth.failure", gateway.events) + } _ = client.Close() <-done } @@ -51,6 +59,66 @@ func upstreamRequestForTest() api.UpstreamConnectRequest { return api.UpstreamConnectRequest{} } +type recordedEvent struct { + name string + fields map[string]string +} + +type recordingGateway struct { + events []recordedEvent + wg sync.WaitGroup +} + +func (g *recordingGateway) HandleConn(net.Conn) {} +func (g *recordingGateway) ExitWaitGroup() *sync.WaitGroup { return &g.wg } +func (g *recordingGateway) Hook(string, any) error { return nil } +func (g *recordingGateway) EmitEvent(_ context.Context, name string, fields map[string]string) error { + g.events = append(g.events, recordedEvent{name: name, fields: fields}) + return nil +} +func (g *recordingGateway) ObserveMetric(context.Context, string, float64, map[string]string) error { + return nil +} +func (g *recordingGateway) Logger() api.Logger { return testLogger{} } +func (g *recordingGateway) DataStore() api.DataStore { return testDataStore{} } +func (g *recordingGateway) FileStore() api.FileStore { return testFileStore{} } +func (g *recordingGateway) ExternalClient(string) api.ExternalClient { return testExternalClient{} } +func (g *recordingGateway) RegisterBackgroundTask(api.BackgroundTask) error { return nil } + +type testLogger struct{} + +func (testLogger) Debug(context.Context, string, map[string]string) {} +func (testLogger) Info(context.Context, string, map[string]string) {} +func (testLogger) Warn(context.Context, string, map[string]string) {} +func (testLogger) Error(context.Context, string, map[string]string) {} + +type testDataStore struct{} + +func (testDataStore) Put(context.Context, api.DataRecord) error { return nil } +func (testDataStore) Get(context.Context, string) (api.DataRecord, error) { + return api.DataRecord{}, nil +} +func (testDataStore) Delete(context.Context, string) error { return nil } + +type testFileStore struct{} + +func (testFileStore) ResourcePath(string) (string, error) { return "", nil } +func (testFileStore) Write(context.Context, string, string, []byte, string, time.Duration) error { + return nil +} +func (testFileStore) Read(context.Context, string, string, int64) ([]byte, error) { return nil, nil } +func (testFileStore) Delete(context.Context, string, string) error { return nil } + +type testExternalClient struct{} + +func (testExternalClient) DoHTTP(context.Context, api.ExternalRequest) (api.ExternalResponse, error) { + return api.ExternalResponse{}, nil +} +func (testExternalClient) DialTCP(context.Context, string, time.Duration) (net.Conn, error) { + return nil, nil +} +func (testExternalClient) HealthCheck(context.Context) error { return nil } + func mcAuthProxyHandshakePacket(host string) []byte { payload := []byte{0x00, 0x63, byte(len(host))} payload = append(payload, host...) diff --git a/examples/plugins/mc-auth-proxy/manifest.json b/examples/plugins/mc-auth-proxy/manifest.json index e08a185..cc446cd 100644 --- a/examples/plugins/mc-auth-proxy/manifest.json +++ b/examples/plugins/mc-auth-proxy/manifest.json @@ -53,11 +53,42 @@ "handler_timeout_ms": 3000, "initial_write_timeout_ms": 1000 }, + "events": [ + { "name": "auth.success", "fields": ["result", "mode"] }, + { "name": "auth.failure", "fields": ["result", "mode"] } + ], + "custom_metrics": [ + { "name": "auth.attempts", "type": "counter", "labels": ["result", "mode"] } + ], + "external_dependencies": [ + { + "name": "backend", + "endpoint": "tcp://", + "purpose": "auth", + "required": true, + "timeout": "3s", + "retry": 0, + "fail_policy": "fail_closed", + "data_classes": ["operational"] + } + ], + "background_tasks": [ + { "id": "profile-cache-gc", "name": "Profile cache GC", "mode": "manual", "manual": true, "timeout": "1s" } + ], + "data_stores": [ + { "name": "profile-cache", "schema_version": 1, "data_class": "profile_cache", "quota_bytes": 1048576, "retention": "24h", "exportable": false } + ], + "file_stores": [ + { "namespace": "cache", "data_class": "profile_cache", "quota_bytes": 1048576, "retention": "24h" }, + { "namespace": "diagnostic", "data_class": "diagnostic", "quota_bytes": 1048576, "retention": "24h" } + ], "config_schema": { "type": "object", "properties": { "match_host": { "type": "string" }, - "message": { "type": "string" } + "fixture_accept": { "type": "boolean" }, + "disconnect_message": { "type": "string" }, + "backend": { "type": "string" } } } } diff --git a/internal/admindb/db.go b/internal/admindb/db.go index 7371b26..a2ead2c 100644 --- a/internal/admindb/db.go +++ b/internal/admindb/db.go @@ -284,6 +284,80 @@ CREATE TABLE IF NOT EXISTS plugin_benchmarks ( created_at INTEGER NOT NULL ); +CREATE TABLE IF NOT EXISTS plugin_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plugin_id TEXT NOT NULL, + name TEXT NOT NULL, + fields_json TEXT NOT NULL DEFAULT '{}', + dropped INTEGER NOT NULL DEFAULT 0, + reason TEXT NOT NULL DEFAULT '', + trace_id TEXT NOT NULL DEFAULT '', + connection_id TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS plugin_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plugin_id TEXT NOT NULL, + level TEXT NOT NULL, + message TEXT NOT NULL, + fields_json TEXT NOT NULL DEFAULT '{}', + trace_id TEXT NOT NULL DEFAULT '', + connection_id TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS plugin_traces ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plugin_id TEXT NOT NULL DEFAULT '', + trace_id TEXT NOT NULL, + connection_id TEXT NOT NULL DEFAULT '', + handler_id TEXT NOT NULL DEFAULT '', + operation TEXT NOT NULL, + status TEXT NOT NULL DEFAULT '', + duration_ms INTEGER NOT NULL DEFAULT 0, + fields_json TEXT NOT NULL DEFAULT '{}', + created_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS plugin_data ( + plugin_id TEXT NOT NULL, + key TEXT NOT NULL, + value BLOB NOT NULL, + schema_version INTEGER NOT NULL DEFAULT 0, + data_class TEXT NOT NULL DEFAULT '', + exportable INTEGER NOT NULL DEFAULT 0, + size_bytes INTEGER NOT NULL DEFAULT 0, + expires_at INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY(plugin_id, key) +); + +CREATE TABLE IF NOT EXISTS plugin_files ( + plugin_id TEXT NOT NULL, + namespace TEXT NOT NULL, + path TEXT NOT NULL, + disk_path TEXT NOT NULL, + data_class TEXT NOT NULL DEFAULT '', + exportable INTEGER NOT NULL DEFAULT 0, + readonly INTEGER NOT NULL DEFAULT 0, + size_bytes INTEGER NOT NULL DEFAULT 0, + expires_at INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY(plugin_id, namespace, path) +); + +CREATE TABLE IF NOT EXISTS plugin_diagnostics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plugin_id TEXT NOT NULL, + path TEXT NOT NULL, + size_bytes INTEGER NOT NULL DEFAULT 0, + sections_json TEXT NOT NULL DEFAULT '[]', + created_at INTEGER NOT NULL +); + CREATE INDEX IF NOT EXISTS idx_routes_enabled ON routes(enabled); CREATE INDEX IF NOT EXISTS idx_audit_logs_created_at ON audit_logs(created_at); CREATE INDEX IF NOT EXISTS idx_plugin_artifacts_plugin_id ON plugin_artifacts(plugin_id, created_at); @@ -299,6 +373,12 @@ CREATE INDEX IF NOT EXISTS idx_plugin_advisories_artifact ON plugin_advisories(a CREATE INDEX IF NOT EXISTS idx_plugin_advisories_plugin ON plugin_advisories(plugin_id, status); CREATE INDEX IF NOT EXISTS idx_plugin_preflight_lookup ON plugin_preflight_results(plugin_id, artifact_id, profile, created_at); CREATE INDEX IF NOT EXISTS idx_plugin_benchmarks_lookup ON plugin_benchmarks(plugin_id, artifact_id, profile, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_events_lookup ON plugin_events(plugin_id, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_logs_lookup ON plugin_logs(plugin_id, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_traces_lookup ON plugin_traces(plugin_id, trace_id, created_at); +CREATE INDEX IF NOT EXISTS idx_plugin_data_expires ON plugin_data(expires_at); +CREATE INDEX IF NOT EXISTS idx_plugin_files_expires ON plugin_files(expires_at); +CREATE INDEX IF NOT EXISTS idx_plugin_diagnostics_lookup ON plugin_diagnostics(plugin_id, created_at); INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES (1, strftime('%s','now')); ` if _, err := db.Exec(schema); err != nil { diff --git a/internal/adminhttp/api.go b/internal/adminhttp/api.go index 526d7b1..b477b1f 100644 --- a/internal/adminhttp/api.go +++ b/internal/adminhttp/api.go @@ -29,22 +29,25 @@ type APIHandlers struct { AuditLogs http.HandlerFunc - PluginArtifacts http.HandlerFunc - PluginArtifact SegmentHandlerFunc - PluginSources http.HandlerFunc - PluginBuilds http.HandlerFunc - PluginBuild SegmentHandlerFunc - PluginGC http.HandlerFunc - PluginsList http.HandlerFunc - PluginItem SegmentHandlerFunc - PluginAction SegmentHandlerFunc - PluginConfig SegmentHandlerFunc - PluginSecrets SegmentHandlerFunc - PluginRollback SegmentHandlerFunc - PluginDraining SegmentHandlerFunc - PluginDispatch http.HandlerFunc - PluginGovernance SegmentHandlerFunc - PluginAdvisories http.HandlerFunc + PluginArtifacts http.HandlerFunc + PluginArtifact SegmentHandlerFunc + PluginSources http.HandlerFunc + PluginBuilds http.HandlerFunc + PluginBuild SegmentHandlerFunc + PluginGC http.HandlerFunc + PluginsList http.HandlerFunc + PluginItem SegmentHandlerFunc + PluginAction SegmentHandlerFunc + PluginConfig SegmentHandlerFunc + PluginSecrets SegmentHandlerFunc + PluginRollback SegmentHandlerFunc + PluginOperations SegmentHandlerFunc + PluginOperationsGC http.HandlerFunc + PluginDraining SegmentHandlerFunc + PluginDispatch http.HandlerFunc + PluginGovernance SegmentHandlerFunc + PluginAdvisories http.HandlerFunc + PluginDiagnostics SegmentHandlerFunc } func NewAPIHandler(prefix string, handlers APIHandlers) http.HandlerFunc { @@ -98,6 +101,8 @@ func NewAPIHandler(prefix string, handlers APIHandlers) http.HandlerFunc { callSegmentHandler(w, r, handlers.PluginBuild, strings.TrimPrefix(path, "/plugin-builds/")) case path == "/plugin-gc" && (r.Method == http.MethodGet || r.Method == http.MethodPost): callHandler(w, r, handlers.PluginGC) + case path == "/plugin-operations-gc" && (r.Method == http.MethodGet || r.Method == http.MethodPost): + callHandler(w, r, handlers.PluginOperationsGC) case path == "/plugins" && r.Method == http.MethodGet: callHandler(w, r, handlers.PluginsList) case path == "/plugins/dispatch-plan" && r.Method == http.MethodGet: @@ -110,6 +115,14 @@ func NewAPIHandler(prefix string, handlers APIHandlers) http.HandlerFunc { callSegmentHandler(w, r, handlers.PluginGovernance, pluginPath) return } + if strings.Contains(pluginPath, "/operations/") || strings.HasSuffix(pluginPath, "/operations") { + callSegmentHandler(w, r, handlers.PluginOperations, pluginPath) + return + } + if strings.HasSuffix(pluginPath, "/diagnostics") { + callSegmentHandler(w, r, handlers.PluginDiagnostics, strings.TrimSuffix(pluginPath, "/diagnostics")) + return + } if strings.HasSuffix(pluginPath, "/draining/force-close") { callSegmentHandler(w, r, handlers.PluginDraining, strings.TrimSuffix(pluginPath, "/draining/force-close")) return diff --git a/internal/adminhttp/api_test.go b/internal/adminhttp/api_test.go index 4c6706a..c1353a9 100644 --- a/internal/adminhttp/api_test.go +++ b/internal/adminhttp/api_test.go @@ -37,6 +37,7 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) { {name: "plugin build", method: http.MethodGet, path: "/admin/api/plugin-builds/7", wantCall: "plugin_build", wantSegment: "7"}, {name: "plugin build retry", method: http.MethodPost, path: "/admin/api/plugin-builds/7/retry", wantCall: "plugin_build", wantSegment: "7/retry"}, {name: "plugin gc", method: http.MethodGet, path: "/admin/api/plugin-gc", wantCall: "plugin_gc"}, + {name: "plugin operations gc", method: http.MethodGet, path: "/admin/api/plugin-operations-gc", wantCall: "plugin_operations_gc"}, {name: "plugins list", method: http.MethodGet, path: "/admin/api/plugins", wantCall: "plugins_list"}, {name: "plugin item", method: http.MethodPut, path: "/admin/api/plugins/upstream-rewrite", wantCall: "plugin_item", wantSegment: "upstream-rewrite"}, {name: "plugin action", method: http.MethodPost, path: "/admin/api/plugins/upstream-rewrite/enable", wantCall: "plugin_action", wantSegment: "upstream-rewrite/enable"}, @@ -46,6 +47,9 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) { {name: "plugin rollback artifact", method: http.MethodPost, path: "/admin/api/plugins/upstream-rewrite/rollback/artifact", wantCall: "plugin_rollback", wantSegment: "upstream-rewrite/rollback/artifact"}, {name: "plugin governance", method: http.MethodGet, path: "/admin/api/plugins/upstream-rewrite/governance", wantCall: "plugin_governance", wantSegment: "upstream-rewrite/governance"}, {name: "plugin governance review", method: http.MethodPost, path: "/admin/api/plugins/upstream-rewrite/governance/review", wantCall: "plugin_governance", wantSegment: "upstream-rewrite/governance/review"}, + {name: "plugin operations", method: http.MethodGet, path: "/admin/api/plugins/upstream-rewrite/operations", wantCall: "plugin_operations", wantSegment: "upstream-rewrite/operations"}, + {name: "plugin operations task trigger", method: http.MethodPost, path: "/admin/api/plugins/upstream-rewrite/operations/tasks/sync/trigger", wantCall: "plugin_operations", wantSegment: "upstream-rewrite/operations/tasks/sync/trigger"}, + {name: "plugin diagnostics", method: http.MethodGet, path: "/admin/api/plugins/upstream-rewrite/diagnostics", wantCall: "plugin_diagnostics", wantSegment: "upstream-rewrite"}, {name: "plugin draining force close", method: http.MethodPost, path: "/admin/api/plugins/mc-auth-proxy/draining/force-close", wantCall: "plugin_draining", wantSegment: "mc-auth-proxy"}, {name: "plugin dispatch", method: http.MethodGet, path: "/admin/api/plugins/dispatch-plan", wantCall: "plugin_dispatch"}, {name: "plugin advisories", method: http.MethodGet, path: "/admin/api/plugin-advisories", wantCall: "plugin_advisories"}, @@ -76,22 +80,25 @@ func TestNewAPIHandlerRoutesRequests(t *testing.T) { AuditLogs: recordCall(&gotCall, "audit_logs"), - PluginArtifacts: recordCall(&gotCall, "plugin_artifacts"), - PluginArtifact: recordSegmentCall(&gotCall, &gotSegment, "plugin_artifact"), - PluginSources: recordCall(&gotCall, "plugin_sources"), - PluginBuilds: recordCall(&gotCall, "plugin_builds"), - PluginBuild: recordSegmentCall(&gotCall, &gotSegment, "plugin_build"), - PluginGC: recordCall(&gotCall, "plugin_gc"), - PluginsList: recordCall(&gotCall, "plugins_list"), - PluginItem: recordSegmentCall(&gotCall, &gotSegment, "plugin_item"), - PluginAction: recordSegmentCall(&gotCall, &gotSegment, "plugin_action"), - PluginConfig: recordSegmentCall(&gotCall, &gotSegment, "plugin_config"), - PluginSecrets: recordSegmentCall(&gotCall, &gotSegment, "plugin_secrets"), - PluginRollback: recordSegmentCall(&gotCall, &gotSegment, "plugin_rollback"), - PluginDraining: recordSegmentCall(&gotCall, &gotSegment, "plugin_draining"), - PluginDispatch: recordCall(&gotCall, "plugin_dispatch"), - PluginGovernance: recordSegmentCall(&gotCall, &gotSegment, "plugin_governance"), - PluginAdvisories: recordCall(&gotCall, "plugin_advisories"), + PluginArtifacts: recordCall(&gotCall, "plugin_artifacts"), + PluginArtifact: recordSegmentCall(&gotCall, &gotSegment, "plugin_artifact"), + PluginSources: recordCall(&gotCall, "plugin_sources"), + PluginBuilds: recordCall(&gotCall, "plugin_builds"), + PluginBuild: recordSegmentCall(&gotCall, &gotSegment, "plugin_build"), + PluginGC: recordCall(&gotCall, "plugin_gc"), + PluginOperationsGC: recordCall(&gotCall, "plugin_operations_gc"), + PluginsList: recordCall(&gotCall, "plugins_list"), + PluginItem: recordSegmentCall(&gotCall, &gotSegment, "plugin_item"), + PluginAction: recordSegmentCall(&gotCall, &gotSegment, "plugin_action"), + PluginConfig: recordSegmentCall(&gotCall, &gotSegment, "plugin_config"), + PluginSecrets: recordSegmentCall(&gotCall, &gotSegment, "plugin_secrets"), + PluginRollback: recordSegmentCall(&gotCall, &gotSegment, "plugin_rollback"), + PluginOperations: recordSegmentCall(&gotCall, &gotSegment, "plugin_operations"), + PluginDraining: recordSegmentCall(&gotCall, &gotSegment, "plugin_draining"), + PluginDispatch: recordCall(&gotCall, "plugin_dispatch"), + PluginGovernance: recordSegmentCall(&gotCall, &gotSegment, "plugin_governance"), + PluginAdvisories: recordCall(&gotCall, "plugin_advisories"), + PluginDiagnostics: recordSegmentCall(&gotCall, &gotSegment, "plugin_diagnostics"), }) resp := httptest.NewRecorder() diff --git a/internal/pluginmanager/artifact.go b/internal/pluginmanager/artifact.go index a800399..f34138e 100644 --- a/internal/pluginmanager/artifact.go +++ b/internal/pluginmanager/artifact.go @@ -21,6 +21,7 @@ import ( var ( pluginIDPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`) secretNamePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`) + schemaKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`) ) type ArtifactStore struct { @@ -108,7 +109,7 @@ func (s ArtifactStore) StoreBuiltBinary(upload ArtifactUpload, manifest Manifest if err != nil { return ArtifactRecord{}, err } - capabilities, err := capabilitiesSummaryJSON(manifest.Capabilities) + capabilities, err := manifestCapabilitiesSummaryJSON(manifest) if err != nil { return ArtifactRecord{}, err } @@ -288,7 +289,7 @@ func (s ArtifactStore) validateAndStore(upload ArtifactUpload, expectedArtifactT if err != nil { return ArtifactRecord{}, err } - capabilities, err := capabilitiesSummaryJSON(manifest.Capabilities) + capabilities, err := manifestCapabilitiesSummaryJSON(manifest) if err != nil { return ArtifactRecord{}, err } @@ -364,7 +365,7 @@ func (s ArtifactStore) storeSourcePackage(upload ArtifactUpload, manifest Manife if err != nil { return ArtifactRecord{}, err } - capabilities, err := capabilitiesSummaryJSON(manifest.Capabilities) + capabilities, err := manifestCapabilitiesSummaryJSON(manifest) if err != nil { return ArtifactRecord{}, err } @@ -427,6 +428,23 @@ func capabilitiesSummaryJSON(raw json.RawMessage) ([]byte, error) { return json.Marshal(summary) } +func manifestCapabilitiesSummaryJSON(manifest Manifest) ([]byte, error) { + data, err := capabilitiesSummaryJSON(manifest.Capabilities) + if err != nil { + return nil, err + } + var summary CapabilitySummary + if err := json.Unmarshal(data, &summary); err != nil { + return nil, err + } + summary.Events = append([]EventSpec(nil), manifest.Events...) + summary.CustomMetrics = append([]MetricSpec(nil), manifest.CustomMetrics...) + summary.ExternalDeps = append([]ExternalSpec(nil), manifest.ExternalDeps...) + summary.DataStores = append([]DataStoreSpec(nil), manifest.DataStores...) + summary.FileStores = append([]FileStoreSpec(nil), manifest.FileStores...) + return json.Marshal(summary) +} + func validateManifest(manifest Manifest) error { switch { case manifest.SchemaVersion != SchemaVersion: @@ -477,9 +495,77 @@ func validateManifest(manifest Manifest) error { } seenSecrets[secret.Name] = true } + if err := validateNamedSpecs("event", eventSpecNames(manifest.Events)); err != nil { + return err + } + if err := validateNamedSpecs("custom metric", metricSpecNames(manifest.CustomMetrics)); err != nil { + return err + } + if err := validateNamedSpecs("background task", taskSpecNames(manifest.BackgroundTasks)); err != nil { + return err + } + if err := validateNamedSpecs("external dependency", externalSpecNames(manifest.ExternalDeps)); err != nil { + return err + } return nil } +func validateNamedSpecs(kind string, names []string) error { + seen := make(map[string]bool, len(names)) + for _, name := range names { + if !schemaKeyPattern.MatchString(name) { + return fmt.Errorf("invalid %s name %q", kind, name) + } + if seen[name] { + return fmt.Errorf("duplicate %s name %q", kind, name) + } + seen[name] = true + } + return nil +} + +func eventSpecNames(specs []EventSpec) []string { + names := make([]string, 0, len(specs)) + for _, spec := range specs { + names = append(names, spec.Name) + for _, field := range spec.Fields { + if !schemaKeyPattern.MatchString(field) { + names = append(names, "invalid field "+field) + } + } + } + return names +} + +func metricSpecNames(specs []MetricSpec) []string { + names := make([]string, 0, len(specs)) + for _, spec := range specs { + names = append(names, spec.Name) + for _, label := range spec.Labels { + if !schemaKeyPattern.MatchString(label) { + names = append(names, "invalid label "+label) + } + } + } + return names +} + +func taskSpecNames(specs []TaskSpec) []string { + names := make([]string, 0, len(specs)) + for _, spec := range specs { + names = append(names, spec.ID) + } + return names +} + +func externalSpecNames(specs []ExternalSpec) []string { + names := make([]string, 0, len(specs)) + for _, spec := range specs { + names = append(names, spec.Name) + } + return names +} + func validateSourceEntries(manifest Manifest, entries map[string]*zip.File) error { if _, ok := entries["go.mod"]; !ok { return errors.New("source package requires go.mod") diff --git a/internal/pluginmanager/manager.go b/internal/pluginmanager/manager.go index 293a620..0b70843 100644 --- a/internal/pluginmanager/manager.go +++ b/internal/pluginmanager/manager.go @@ -138,6 +138,7 @@ type Manager struct { proxySeq uint64 proxyConns map[uint64]*proxyConnection drainingIDs map[string]bool + operations *Operations } type loadedPlugin struct { @@ -171,6 +172,9 @@ type upstreamHandler struct { proxyBytesIn atomic.Uint64 proxyBytesOut atomic.Uint64 proxyDuration atomic.Uint64 + durationCount atomic.Uint64 + durationSumMS atomic.Uint64 + durationMaxMS atomic.Uint64 } type proxyConnection struct { @@ -224,6 +228,7 @@ func New(options Options) *Manager { proxyConns: make(map[uint64]*proxyConnection), drainingIDs: make(map[string]bool), } + manager.operations = NewOperations(manager.repo, options.ArtifactRoot) if manager.builders == nil { manager.builders = map[string]SourceBuilder{ BuilderTypeLocalProcess: LocalProcessBuilder{StoreRoot: options.ArtifactRoot}, @@ -733,6 +738,7 @@ func (m *Manager) Disable(ctx context.Context, actor, pluginID string) (PluginRe } m.removeFromDispatchLocked(pluginID) m.markDrainingLocked(pluginID) + m.operations.StopPlugin(pluginID) if loaded := m.loaded[pluginID]; loaded != nil && loaded.instance != nil { if err := loaded.instance.Destroy(); err != nil { _ = m.repo.RecordOperation(ctx, pluginID, loaded.artifact.ID, "disable", "warning", actor, err.Error(), nil) @@ -762,6 +768,7 @@ func (m *Manager) Delete(ctx context.Context, actor, pluginID string) error { } m.removeFromDispatchLocked(pluginID) m.markDrainingLocked(pluginID) + m.operations.StopPlugin(pluginID) if loaded := m.loaded[pluginID]; loaded != nil && loaded.instance != nil { _ = loaded.instance.Destroy() } @@ -835,7 +842,26 @@ func (m *Manager) ConnectUpstream(ctx context.Context, req api.UpstreamConnectRe if !accepted { continue } + req.Context = WithTraceContext(req.Context, handler.pluginID, req.TraceID, req.ConnectionID, handler.handlerID) + start := time.Now() conn, err := handler.invoke(req) + status := "ok" + if err != nil { + status = "error" + } + _ = m.repo.SaveTrace(context.Background(), TraceSummary{ + PluginID: handler.pluginID, + TraceID: req.TraceID, + ConnectionID: req.ConnectionID, + HandlerID: handler.handlerID, + Operation: "plugin.handler." + handler.handlerID, + Status: status, + DurationMS: time.Since(start).Milliseconds(), + }, map[string]string{ + "host": req.ServerHost, + "upstream": req.UpstreamAddress, + "mode": handler.mode, + }) if errors.Is(err, api.ErrPass) { continue } @@ -843,6 +869,17 @@ func (m *Manager) ConnectUpstream(ctx context.Context, req api.UpstreamConnectRe return UpstreamResult{Handled: true}, err } if conn != nil { + if handler.mode == UpstreamModeDialer { + _ = m.repo.SaveTrace(context.Background(), TraceSummary{ + PluginID: handler.pluginID, + TraceID: req.TraceID, + ConnectionID: req.ConnectionID, + HandlerID: handler.handlerID, + Operation: "backend.dial", + Status: "plugin_supplied", + DurationMS: 0, + }, map[string]string{"upstream": req.UpstreamAddress}) + } result := UpstreamResult{ Conn: conn, Handled: true, @@ -1144,6 +1181,73 @@ func (m *Manager) DispatchPlan(ctx context.Context) DispatchPlan { return plan } +func (m *Manager) OperationsSnapshot(ctx context.Context, pluginID string) (OperationsSnapshot, error) { + if pluginID != "" { + if _, err := m.repo.Plugin(ctx, pluginID); err != nil { + return OperationsSnapshot{}, err + } + } + plan := m.DispatchPlan(ctx) + var handlers []DispatchHandlerSummary + for _, handler := range plan.Handlers { + if pluginID == "" || handler.PluginID == pluginID { + handlers = append(handlers, handler) + } + } + builds, err := m.repo.ListBuilds(ctx, pluginID) + if err != nil { + return OperationsSnapshot{}, err + } + gc, _ := m.operations.GCCandidates(ctx, pluginID) + return m.operations.Snapshot(ctx, pluginID, handlers, builds, gc), nil +} + +func (m *Manager) TriggerBackgroundTask(ctx context.Context, actor, pluginID, taskID, confirmToken string) (BackgroundTaskSummary, error) { + summary, err := m.operations.TriggerTask(pluginID, taskID, confirmToken) + if err != nil { + _ = m.repo.RecordOperation(ctx, pluginID, "", "background_task_trigger", "failed", actor, err.Error(), map[string]any{"task_id": taskID}) + return BackgroundTaskSummary{}, err + } + _ = m.repo.RecordOperation(ctx, pluginID, "", "background_task_trigger", "succeeded", actor, "background task triggered", map[string]any{"task_id": taskID}) + return summary, nil +} + +func (m *Manager) DiagnosticPackage(ctx context.Context, actor, pluginID string) ([]byte, DiagnosticPackageSummary, error) { + plugin, err := m.repo.Plugin(ctx, pluginID) + if err != nil { + return nil, DiagnosticPackageSummary{}, err + } + artifact, err := m.repo.Artifact(ctx, plugin.DesiredArtifactID) + if err != nil { + return nil, DiagnosticPackageSummary{}, err + } + var manifest Manifest + _ = json.Unmarshal([]byte(artifact.MetadataJSON), &manifest) + plan := m.DispatchPlan(ctx) + var handlers []DispatchHandlerSummary + for _, handler := range plan.Handlers { + if handler.PluginID == pluginID { + handlers = append(handlers, handler) + } + } + builds, _ := m.repo.ListBuilds(ctx, pluginID) + gc, _ := m.operations.GCCandidates(ctx, pluginID) + data, summary, err := m.operations.DiagnosticPackage(ctx, plugin, manifest, handlers, builds, gc) + if err != nil { + _ = m.repo.RecordOperation(ctx, pluginID, artifact.ID, "diagnostic_package", "failed", actor, err.Error(), nil) + return nil, DiagnosticPackageSummary{}, err + } + _ = m.repo.RecordOperation(ctx, pluginID, artifact.ID, "diagnostic_package", "succeeded", actor, "diagnostic package generated", map[string]any{ + "size_bytes": summary.SizeBytes, + "sections": summary.Sections, + }) + return data, summary, nil +} + +func (m *Manager) RunOperationsGC(ctx context.Context, actor, pluginID string, dryRun bool) ([]GCCandidate, error) { + return m.operations.RunGC(ctx, actor, pluginID, dryRun) +} + func (m *Manager) TrackProxyConnection(result UpstreamResult, client, endpoint net.Conn) *ProxyConnectionHandle { if result.Mode != UpstreamModeProtocolProxy || client == nil || endpoint == nil { return nil @@ -1258,7 +1362,11 @@ func (m *Manager) loadLocked(ctx context.Context, pluginRecord PluginRecord) (*l return nil, err } - gateway := NewGateway(pluginRecord.ID, m.handleConn, m.wg) + var manifest Manifest + if err := json.Unmarshal([]byte(artifact.MetadataJSON), &manifest); err != nil { + return nil, err + } + gateway := NewGateway(pluginRecord.ID, m.handleConn, m.wg, m.operations.ForPlugin(pluginRecord.ID, artifact.ID, manifest)) instance, err := m.adapter.Load(ctx, artifact, pluginRecord, gateway) if err != nil { _ = m.repo.MarkRuntime(ctx, pluginRecord.ID, RuntimeFailed, "", "", pluginRecord.AppliedGeneration, err.Error(), map[string]any{"error": err.Error()}, nil) @@ -1278,6 +1386,7 @@ func (m *Manager) loadLocked(ctx context.Context, pluginRecord PluginRecord) (*l }, handlerSummaries(handlers)); err != nil { return nil, err } + m.operations.StartTasks(pluginRecord.ID) return loaded, nil } @@ -1326,6 +1435,7 @@ func (m *Manager) restartRequired(pluginID, artifactID string) bool { } func (m *Manager) markEnabled(ctx context.Context, loaded *loadedPlugin) error { + m.operations.StartTasks(loaded.record.ID) return m.repo.MarkRuntime(ctx, loaded.record.ID, RuntimeEnabled, loaded.artifact.ID, loaded.artifact.ID, loaded.record.DesiredGeneration, "", map[string]any{ "handler_count": len(loaded.handlers), }, handlerSummaries(loaded.handlers)) @@ -1454,6 +1564,18 @@ func upstreamModeFromArtifact(artifact ArtifactRecord) string { func (h *upstreamHandler) invoke(req api.UpstreamConnectRequest) (conn net.Conn, err error) { h.calls.Add(1) + start := time.Now() + defer func() { + durationMS := uint64(time.Since(start).Milliseconds()) + h.durationCount.Add(1) + h.durationSumMS.Add(durationMS) + for { + current := h.durationMaxMS.Load() + if durationMS <= current || h.durationMaxMS.CompareAndSwap(current, durationMS) { + break + } + } + }() ctx := req.Context if ctx == nil { ctx = context.Background() @@ -1966,6 +2088,9 @@ func handlerSummaries(handlers []*upstreamHandler) []DispatchHandlerSummary { ProxyBytesIn: handler.proxyBytesIn.Load(), ProxyBytesOut: handler.proxyBytesOut.Load(), ProxyDurationMS: handler.proxyDuration.Load(), + DurationCount: handler.durationCount.Load(), + DurationSumMS: handler.durationSumMS.Load(), + DurationMaxMS: handler.durationMaxMS.Load(), }) } return summaries diff --git a/internal/pluginmanager/manager_test.go b/internal/pluginmanager/manager_test.go index 4178782..1f95c9e 100644 --- a/internal/pluginmanager/manager_test.go +++ b/internal/pluginmanager/manager_test.go @@ -6,6 +6,7 @@ import ( "database/sql" "encoding/json" "errors" + "fmt" "io" "net" "path/filepath" @@ -197,6 +198,154 @@ func TestManagerSecretVersionsAreSummariesOnly(t *testing.T) { } } +func TestManagerOperationsRecordsHandlerMetricsEventsAndDiagnostics(t *testing.T) { + var gateway *Gateway + manager := newManagerForTest(t, &fakeAdapter{ + init: func(g *Gateway) { + gateway = g + }, + handlers: map[string]api.UpstreamConnectHandler{ + "plugin-a": func(req api.UpstreamConnectRequest) (net.Conn, error) { + _ = gateway.EmitEvent(req.Context, "auth.success", map[string]string{"result": "fixture_accept", "mode": "fixture"}) + gateway.Logger().Info(req.Context, "auth success token=secret-value", map[string]string{"result": "fixture_accept"}) + return newMemoryConn(), nil + }, + }, + }) + artifact := uploadTestArtifactWithManifest(t, manager, "plugin-a", func(manifest *Manifest) { + manifest.Events = []EventSpec{{Name: "auth.success", Fields: []string{"result", "mode"}}} + manifest.CustomMetrics = []MetricSpec{{Name: "auth.attempts", Type: "counter", Labels: []string{"result", "mode"}}} + }) + if _, err := manager.SetDesired(context.Background(), "admin", "plugin-a", artifact.ID, DesiredEnabled, `{}`, 10); err != nil { + t.Fatalf("SetDesired() error = %v", err) + } + if _, err := manager.Enable(context.Background(), "admin", "plugin-a"); err != nil { + t.Fatalf("Enable() error = %v", err) + } + if _, err := manager.ConnectUpstream(context.Background(), api.UpstreamConnectRequest{Host: "play.example", Upstream: "backend"}); err != nil { + t.Fatalf("ConnectUpstream() error = %v", err) + } + deadline := time.Now().Add(time.Second) + var snap OperationsSnapshot + for time.Now().Before(deadline) { + var err error + snap, err = manager.OperationsSnapshot(context.Background(), "plugin-a") + if err != nil { + t.Fatalf("OperationsSnapshot() error = %v", err) + } + if len(snap.Events) > 0 && len(snap.Handlers) > 0 && snap.Handlers[0].Calls > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + if len(snap.Events) == 0 || snap.Events[0].Name != "auth.success" { + t.Fatalf("events = %+v, want auth.success", snap.Events) + } + if len(snap.Handlers) == 0 || snap.Handlers[0].Calls == 0 || snap.Handlers[0].DurationCount == 0 { + t.Fatalf("handler metrics = %+v, want calls and duration", snap.Handlers) + } + data, _, err := manager.DiagnosticPackage(context.Background(), "admin", "plugin-a") + if err != nil { + t.Fatalf("DiagnosticPackage() error = %v", err) + } + if bytes.Contains(data, []byte("secret-value")) || bytes.Contains(data, []byte("packet")) { + t.Fatalf("diagnostic leaked sensitive content: %s", data) + } +} + +func TestManagerOperationsRejectsUndeclaredAndHighCardinalityEvents(t *testing.T) { + var gateway *Gateway + manager := newManagerForTest(t, &fakeAdapter{init: func(g *Gateway) { gateway = g }}) + artifact := uploadTestArtifactWithManifest(t, manager, "plugin-a", func(manifest *Manifest) { + manifest.Events = []EventSpec{{Name: "auth.failure", Fields: []string{"result"}}} + }) + if _, err := manager.SetDesired(context.Background(), "admin", "plugin-a", artifact.ID, DesiredEnabled, `{}`, 10); err != nil { + t.Fatalf("SetDesired() error = %v", err) + } + if _, err := manager.Enable(context.Background(), "admin", "plugin-a"); err != nil { + t.Fatalf("Enable() error = %v", err) + } + if err := gateway.EmitEvent(context.Background(), "auth.success", map[string]string{"result": "ok"}); err == nil { + t.Fatal("EmitEvent(undeclared) error = nil") + } + var highCardinalityErr error + for i := 0; i < 70; i++ { + highCardinalityErr = gateway.EmitEvent(context.Background(), "auth.failure", map[string]string{"result": fmt.Sprintf("result-%02d", i)}) + if highCardinalityErr != nil { + break + } + } + if highCardinalityErr == nil { + t.Fatal("EmitEvent(high-cardinality) error = nil") + } +} + +func TestManagerOperationsBackgroundTaskDataQuotaExternalAndGC(t *testing.T) { + var gateway *Gateway + manager := newManagerForTest(t, &fakeAdapter{init: func(g *Gateway) { + gateway = g + _ = g.RegisterBackgroundTask(api.BackgroundTask{ + ID: "sync", + Manual: true, + Timeout: 20 * time.Millisecond, + Run: func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + }, + }) + }}) + artifact := uploadTestArtifactWithManifest(t, manager, "plugin-a", func(manifest *Manifest) { + manifest.BackgroundTasks = []TaskSpec{{ID: "sync", Mode: "manual", Manual: true, Timeout: "20ms"}} + manifest.DataStores = []DataStoreSpec{{Name: "cache", SchemaVersion: 1, QuotaBytes: 8, DataClass: "cache"}} + manifest.ExternalDeps = []ExternalSpec{{Name: "session", Endpoint: "http://127.0.0.1:1", Purpose: "auth", Timeout: "20ms", Required: true}} + }) + if _, err := manager.SetDesired(context.Background(), "admin", "plugin-a", artifact.ID, DesiredEnabled, `{}`, 10); err != nil { + t.Fatalf("SetDesired() error = %v", err) + } + if _, err := manager.Enable(context.Background(), "admin", "plugin-a"); err != nil { + t.Fatalf("Enable() error = %v", err) + } + start := time.Now() + if _, err := manager.TriggerBackgroundTask(context.Background(), "admin", "plugin-a", "sync", manager.operations.plugins["plugin-a"].tasks["sync"].confirmToken); err != nil { + t.Fatalf("TriggerBackgroundTask() error = %v", err) + } + if time.Since(start) > time.Second { + t.Fatal("background task trigger blocked too long") + } + if err := gateway.DataStore().Put(context.Background(), api.DataRecord{Key: "one", Value: []byte("12345678"), SchemaVersion: 1}); err != nil { + t.Fatalf("DataStore.Put(within quota) error = %v", err) + } + if err := gateway.DataStore().Put(context.Background(), api.DataRecord{Key: "two", Value: []byte("x")}); err == nil { + t.Fatal("DataStore.Put(over quota) error = nil") + } + if _, err := gateway.ExternalClient("session").DoHTTP(context.Background(), api.ExternalRequest{Method: "GET", URL: "http://127.0.0.1:1"}); err == nil { + t.Fatal("ExternalClient.DoHTTP() error = nil, want connection error") + } + snap, err := manager.OperationsSnapshot(context.Background(), "plugin-a") + if err != nil { + t.Fatalf("OperationsSnapshot() error = %v", err) + } + if len(snap.ExternalDependencies) == 0 || snap.ExternalDependencies[0].Errors == 0 { + t.Fatalf("external summaries = %+v, want error count", snap.ExternalDependencies) + } + if err := manager.repo.PutPluginData(context.Background(), PluginDataSummary{PluginID: "plugin-a", Key: "expired", SizeBytes: 3, ExpiresAt: time.Now().Add(-time.Second).Unix()}, []byte("old")); err != nil { + t.Fatalf("PutPluginData(expired) error = %v", err) + } + candidates, err := manager.RunOperationsGC(context.Background(), "admin", "plugin-a", true) + if err != nil { + t.Fatalf("RunOperationsGC(dry-run) error = %v", err) + } + found := false + for _, candidate := range candidates { + if candidate.Kind == "plugin_data" && candidate.ID == "expired" && candidate.SizeBytes > 0 { + found = true + } + } + if !found { + t.Fatalf("gc candidates = %+v, want expired plugin_data with size", candidates) + } +} + func TestManagerRollbackConfigSnapshotRunsDryRunBeforeChangingDesired(t *testing.T) { adapter := &fakeAdapter{} manager := newManagerForTest(t, adapter) @@ -905,6 +1054,7 @@ func waitForPluginManagerTest(t *testing.T, done func() bool) { type fakeAdapter struct { loads int handlers map[string]api.UpstreamConnectHandler + init func(*Gateway) loadErr error loadErrs map[string]error dryRunErr error @@ -933,6 +1083,9 @@ func (a *fakeAdapter) Load(_ context.Context, artifact ArtifactRecord, _ PluginR ); err != nil { return nil, err } + if a.init != nil { + a.init(gateway) + } return &fakePlugin{}, nil } diff --git a/internal/pluginmanager/operations.go b/internal/pluginmanager/operations.go new file mode 100644 index 0000000..94b2df8 --- /dev/null +++ b/internal/pluginmanager/operations.go @@ -0,0 +1,1590 @@ +package pluginmanager + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "os" + "path" + "path/filepath" + "regexp" + "sort" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/rs/zerolog/log" + "github.com/tursom/mc-gateway/plugin/api" +) + +const ( + circuitClosed = "closed" + circuitOpen = "open" +) + +var ( + metricNamePattern = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_.:-]{0,127}$`) + storeKeyPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_.:/-]{0,255}$`) +) + +type traceContextKey struct{} + +type traceContext struct { + PluginID string + TraceID string + ConnectionID string + HandlerID string +} + +type Operations struct { + repo Repository + root string + + mu sync.RWMutex + plugins map[string]*PluginOperations + + eventQueue chan queuedEvent + queued atomic.Uint64 + dropped atomic.Uint64 + deadLetter atomic.Uint64 +} + +type queuedEvent struct { + pluginID string + name string + fields map[string]string + dropped bool + reason string + traceID string + connectionID string +} + +type PluginOperations struct { + parent *Operations + pluginID string + artifactID string + manifest Manifest + + mu sync.Mutex + eventSchemas map[string]map[string]bool + metricSchemas map[string]MetricSpec + taskSpecs map[string]TaskSpec + externalSpecs map[string]ExternalSpec + dataQuota int64 + dataSpec DataStoreSpec + fileQuota int64 + fileSpecs map[string]FileStoreSpec + eventValues map[string]map[string]map[string]struct{} + eventSummaries map[string]*EventSummary + metricSummaries map[string]*CustomMetricSummary + externals map[string]*externalRuntime + tasks map[string]*taskRuntime +} + +type externalRuntime struct { + spec ExternalSpec + + requests atomic.Uint64 + errors atomic.Uint64 + inflight atomic.Int64 + durationCount atomic.Uint64 + durationSumMS atomic.Uint64 + consecutiveFailures atomic.Uint64 + + mu sync.Mutex + circuitUntil time.Time + recentError string + lastStatus string + lastSeenAt int64 +} + +type taskRuntime struct { + pluginID string + spec TaskSpec + task api.BackgroundTask + confirmToken string + + mu sync.Mutex + cancel context.CancelFunc + running bool + schedulerOn bool + lastRunAt int64 + nextRunAt int64 + lastDurationMS int64 + lastError string + skipped uint64 + consecutiveFailures uint64 +} + +func NewOperations(repo Repository, root string) *Operations { + ops := &Operations{ + repo: repo, + root: root, + plugins: make(map[string]*PluginOperations), + eventQueue: make(chan queuedEvent, DefaultEventQueueLimit), + } + go ops.consumeEvents() + return ops +} + +func (o *Operations) ForPlugin(pluginID, artifactID string, manifest Manifest) *PluginOperations { + o.mu.Lock() + defer o.mu.Unlock() + po := o.plugins[pluginID] + if po == nil { + po = &PluginOperations{ + parent: o, + pluginID: pluginID, + eventValues: make(map[string]map[string]map[string]struct{}), + eventSummaries: make(map[string]*EventSummary), + metricSummaries: make(map[string]*CustomMetricSummary), + externals: make(map[string]*externalRuntime), + tasks: make(map[string]*taskRuntime), + } + o.plugins[pluginID] = po + } + po.configure(artifactID, manifest) + return po +} + +func (o *Operations) StopPlugin(pluginID string) { + o.mu.RLock() + po := o.plugins[pluginID] + o.mu.RUnlock() + if po != nil { + po.stopTasks() + } +} + +func (o *Operations) StartTasks(pluginID string) { + o.mu.RLock() + po := o.plugins[pluginID] + o.mu.RUnlock() + if po != nil { + po.StartTasks(pluginID) + } +} + +func (o *Operations) consumeEvents() { + for event := range o.eventQueue { + o.queued.Add(1) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + err := o.repo.SaveEvent(ctx, EventSummary{ + PluginID: event.pluginID, + Name: event.name, + Fields: event.fields, + }, event.dropped, event.reason, event.traceID, event.connectionID) + cancel() + if err != nil { + o.deadLetter.Add(1) + } + } +} + +func (o *Operations) queueEvent(event queuedEvent) { + select { + case o.eventQueue <- event: + default: + o.dropped.Add(1) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + _ = o.repo.SaveEvent(ctx, EventSummary{ + PluginID: event.pluginID, + Name: event.name, + Fields: event.fields, + }, true, "event_queue_full", event.traceID, event.connectionID) + cancel() + } +} + +func (po *PluginOperations) configure(artifactID string, manifest Manifest) { + po.mu.Lock() + defer po.mu.Unlock() + po.artifactID = artifactID + po.manifest = manifest + + po.eventSchemas = make(map[string]map[string]bool) + for _, spec := range manifest.Events { + if spec.Name == "" { + continue + } + fields := make(map[string]bool, len(spec.Fields)) + for _, field := range spec.Fields { + fields[field] = true + } + po.eventSchemas[spec.Name] = fields + } + po.metricSchemas = make(map[string]MetricSpec) + for _, spec := range manifest.CustomMetrics { + if spec.Name != "" { + po.metricSchemas[spec.Name] = spec + } + } + po.taskSpecs = make(map[string]TaskSpec) + for _, spec := range manifest.BackgroundTasks { + if spec.ID != "" { + po.taskSpecs[spec.ID] = spec + } + } + po.externalSpecs = make(map[string]ExternalSpec) + for _, spec := range manifest.ExternalDeps { + if spec.Name != "" { + po.externalSpecs[spec.Name] = spec + if po.externals[spec.Name] == nil { + po.externals[spec.Name] = &externalRuntime{spec: spec} + } else { + po.externals[spec.Name].spec = spec + } + } + } + po.dataQuota = DefaultPluginDataQuota + po.dataSpec = DataStoreSpec{QuotaBytes: DefaultPluginDataQuota} + if len(manifest.DataStores) > 0 { + po.dataSpec = manifest.DataStores[0] + if po.dataSpec.QuotaBytes > 0 { + po.dataQuota = po.dataSpec.QuotaBytes + } + } + po.fileQuota = DefaultPluginFileQuota + po.fileSpecs = make(map[string]FileStoreSpec) + for _, spec := range manifest.FileStores { + if spec.Namespace != "" { + po.fileSpecs[spec.Namespace] = spec + if spec.QuotaBytes > 0 && spec.QuotaBytes < po.fileQuota { + po.fileQuota = spec.QuotaBytes + } + } + } + if len(po.fileSpecs) == 0 { + for _, namespace := range []string{"data", "cache", "tmp", "log", "diagnostic"} { + po.fileSpecs[namespace] = FileStoreSpec{Namespace: namespace, QuotaBytes: DefaultPluginFileQuota} + } + } +} + +func (po *PluginOperations) EmitEvent(ctx context.Context, name string, fields map[string]string) error { + trace := traceFromContext(ctx) + clean, dropReason, err := po.validateEvent(name, fields) + if err != nil { + po.parent.queueEvent(queuedEvent{ + pluginID: po.pluginID, name: name, fields: clean, dropped: true, + reason: dropReason, traceID: trace.TraceID, connectionID: trace.ConnectionID, + }) + return err + } + po.mu.Lock() + summary := po.eventSummaries[name] + if summary == nil { + summary = &EventSummary{PluginID: po.pluginID, Name: name} + po.eventSummaries[name] = summary + } + summary.Count++ + summary.Fields = clean + summary.LastSeenAt = time.Now().Unix() + po.mu.Unlock() + po.parent.queueEvent(queuedEvent{ + pluginID: po.pluginID, name: name, fields: clean, + traceID: trace.TraceID, connectionID: trace.ConnectionID, + }) + return nil +} + +func (po *PluginOperations) ObserveMetric(ctx context.Context, name string, value float64, labels map[string]string) error { + _ = ctx + clean, metricType, err := po.validateMetric(name, labels) + if err != nil { + return err + } + po.mu.Lock() + defer po.mu.Unlock() + summary := po.metricSummaries[name] + if summary == nil { + summary = &CustomMetricSummary{PluginID: po.pluginID, Name: name, Type: metricType} + po.metricSummaries[name] = summary + } + summary.Count++ + summary.LastValue = value + summary.Labels = clean + summary.LastSeenAt = time.Now().Unix() + return nil +} + +func (po *PluginOperations) Logger() api.Logger { + return pluginLogger{ops: po} +} + +func (po *PluginOperations) DataStore() api.DataStore { + return pluginDataStore{ops: po} +} + +func (po *PluginOperations) FileStore() api.FileStore { + return pluginFileStore{ops: po} +} + +func (po *PluginOperations) ExternalClient(name string) api.ExternalClient { + po.mu.Lock() + defer po.mu.Unlock() + runtime := po.externals[name] + if runtime == nil { + spec := po.externalSpecs[name] + runtime = &externalRuntime{spec: spec} + po.externals[name] = runtime + } + return pluginExternalClient{ops: po, name: name, runtime: runtime} +} + +func (po *PluginOperations) RegisterBackgroundTask(task api.BackgroundTask) error { + if task.ID == "" { + return errors.New("background task id is required") + } + if !metricNamePattern.MatchString(task.ID) { + return fmt.Errorf("invalid background task id %q", task.ID) + } + if task.Run == nil { + return fmt.Errorf("background task %q run function is required", task.ID) + } + po.mu.Lock() + defer po.mu.Unlock() + spec := po.taskSpecs[task.ID] + if len(po.taskSpecs) > 0 && spec.ID == "" { + return fmt.Errorf("background task %q is not declared by manifest", task.ID) + } + if spec.ID == "" { + spec = TaskSpec{ID: task.ID, Name: task.Name} + } + if task.Name == "" { + task.Name = spec.Name + } + if task.Interval == 0 && spec.Interval != "" { + task.Interval, _ = time.ParseDuration(spec.Interval) + } + if task.Timeout == 0 && spec.Timeout != "" { + task.Timeout, _ = time.ParseDuration(spec.Timeout) + } + if task.Jitter == 0 && spec.Jitter != "" { + task.Jitter, _ = time.ParseDuration(spec.Jitter) + } + if !task.RunOnStart { + task.RunOnStart = spec.RunOnStart + } + if !task.Manual { + task.Manual = spec.Manual || spec.Mode == "manual" + } + if task.Timeout <= 0 { + task.Timeout = DefaultHandlerTimeout + } + rt := po.tasks[task.ID] + if rt == nil { + rt = &taskRuntime{ + pluginID: po.pluginID, + spec: spec, + confirmToken: randomToken(), + } + po.tasks[task.ID] = rt + } + rt.task = task + rt.spec = spec + return nil +} + +func (po *PluginOperations) StartTasks(pluginID string) { + if pluginID != po.pluginID { + return + } + po.mu.Lock() + tasks := make([]*taskRuntime, 0, len(po.tasks)) + for _, task := range po.tasks { + tasks = append(tasks, task) + } + po.mu.Unlock() + for _, task := range tasks { + po.startTask(task) + } +} + +func (po *PluginOperations) stopTasks() { + po.mu.Lock() + tasks := make([]*taskRuntime, 0, len(po.tasks)) + for _, task := range po.tasks { + tasks = append(tasks, task) + } + po.mu.Unlock() + for _, task := range tasks { + task.mu.Lock() + if task.cancel != nil { + task.cancel() + task.cancel = nil + } + task.schedulerOn = false + task.nextRunAt = 0 + task.mu.Unlock() + } +} + +func (po *PluginOperations) startTask(task *taskRuntime) { + task.mu.Lock() + if task.schedulerOn { + task.mu.Unlock() + return + } + task.schedulerOn = true + interval := task.task.Interval + runOnStart := task.task.RunOnStart + task.mu.Unlock() + + if runOnStart { + go po.runTask(task) + } + if interval <= 0 || task.task.Manual { + return + } + go func() { + for { + delay := interval + deterministicJitter(task.task.Jitter, task.task.ID) + task.mu.Lock() + if !task.schedulerOn { + task.mu.Unlock() + return + } + task.nextRunAt = time.Now().Add(delay).Unix() + task.mu.Unlock() + timer := time.NewTimer(delay) + <-timer.C + task.mu.Lock() + on := task.schedulerOn + task.mu.Unlock() + if !on { + return + } + po.runTask(task) + } + }() +} + +func (po *PluginOperations) runTask(task *taskRuntime) { + task.mu.Lock() + if task.running { + task.skipped++ + task.mu.Unlock() + return + } + task.running = true + timeout := task.task.Timeout + if timeout <= 0 { + timeout = DefaultHandlerTimeout + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + task.cancel = cancel + task.mu.Unlock() + + start := time.Now() + err := task.task.Run(ctx) + cancel() + + task.mu.Lock() + task.running = false + task.cancel = nil + task.lastRunAt = start.Unix() + task.lastDurationMS = time.Since(start).Milliseconds() + if err != nil { + task.consecutiveFailures++ + task.lastError = redactSensitive(err.Error()) + } else { + task.consecutiveFailures = 0 + task.lastError = "" + } + task.mu.Unlock() +} + +func (po *PluginOperations) TriggerTask(taskID, confirmToken string) (BackgroundTaskSummary, error) { + po.mu.Lock() + task := po.tasks[taskID] + po.mu.Unlock() + if task == nil { + return BackgroundTaskSummary{}, fmt.Errorf("background task %q not found", taskID) + } + task.mu.Lock() + expected := task.confirmToken + task.mu.Unlock() + if expected == "" || confirmToken != expected { + return BackgroundTaskSummary{}, errors.New("confirm_token is required") + } + go po.runTask(task) + return task.summary(), nil +} + +func (po *PluginOperations) Snapshot(ctx context.Context, pluginID string, handlers []DispatchHandlerSummary, builds []BuildRecord, gc []GCCandidate) OperationsSnapshot { + _ = ctx + po.mu.Lock() + events := make([]EventSummary, 0, len(po.eventSummaries)) + for _, event := range po.eventSummaries { + events = append(events, *event) + } + metrics := make([]CustomMetricSummary, 0, len(po.metricSummaries)) + for _, metric := range po.metricSummaries { + metrics = append(metrics, *metric) + } + externals := make([]ExternalDependencySummary, 0, len(po.externals)) + for name, ext := range po.externals { + externals = append(externals, ext.summary(po.pluginID, name)) + } + tasks := make([]BackgroundTaskSummary, 0, len(po.tasks)) + for _, task := range po.tasks { + tasks = append(tasks, task.summary()) + } + po.mu.Unlock() + + recentEvents, _ := po.parent.repo.RecentEvents(ctx, pluginID, DefaultEventRecentLimit) + if len(recentEvents) > 0 { + events = mergeEventSummaries(events, recentEvents) + } + logs, _ := po.parent.repo.RecentLogs(ctx, pluginID, DefaultLogRecentLimit) + traces, _ := po.parent.repo.RecentTraces(ctx, pluginID, 200) + data, _ := po.parent.repo.ListPluginData(ctx, pluginID) + files, _ := po.parent.repo.ListPluginFiles(ctx, pluginID) + diagnostics, _ := po.parent.repo.ListDiagnostics(ctx, pluginID, 20) + + buildMetrics := make([]BuildMetricSummary, 0, len(builds)) + for _, build := range builds { + buildMetrics = append(buildMetrics, BuildMetricSummary{ + PluginID: build.PluginID, + BuildID: build.ID, + Status: build.Status, + DurationMS: build.DurationMS, + Failed: build.Status == BuildStatusFailed, + ErrorRedacted: redactSensitive(build.Error), + CreatedAt: build.CreatedAt, + }) + } + sort.Slice(events, func(i, j int) bool { return events[i].LastSeenAt > events[j].LastSeenAt }) + sort.Slice(metrics, func(i, j int) bool { return metrics[i].LastSeenAt > metrics[j].LastSeenAt }) + sort.Slice(externals, func(i, j int) bool { return externals[i].Name < externals[j].Name }) + return OperationsSnapshot{ + PluginID: pluginID, + UpdatedAt: time.Now().Unix(), + Handlers: handlers, + Builds: buildMetrics, + Events: events, + CustomMetrics: metrics, + Logs: logs, + Traces: traces, + BackgroundTasks: tasks, + PluginData: data, + PluginFiles: files, + ExternalDependencies: externals, + GC: gc, + EventQueue: EventQueueSummary{ + Limit: DefaultEventQueueLimit, + Queued: len(po.parent.eventQueue), + Dropped: po.parent.dropped.Load(), + DeadLetters: po.parent.deadLetter.Load(), + }, + Diagnostics: diagnostics, + } +} + +func (o *Operations) Snapshot(ctx context.Context, pluginID string, handlers []DispatchHandlerSummary, builds []BuildRecord, gc []GCCandidate) OperationsSnapshot { + o.mu.RLock() + po := o.plugins[pluginID] + o.mu.RUnlock() + if po == nil { + po = &PluginOperations{parent: o, pluginID: pluginID} + } + return po.Snapshot(ctx, pluginID, handlers, builds, gc) +} + +func (o *Operations) TriggerTask(pluginID, taskID, confirmToken string) (BackgroundTaskSummary, error) { + o.mu.RLock() + po := o.plugins[pluginID] + o.mu.RUnlock() + if po == nil { + return BackgroundTaskSummary{}, ErrPluginNotFound + } + return po.TriggerTask(taskID, confirmToken) +} + +func (o *Operations) DiagnosticPackage(ctx context.Context, plugin PluginRecord, manifest Manifest, handlers []DispatchHandlerSummary, builds []BuildRecord, gc []GCCandidate) ([]byte, DiagnosticPackageSummary, error) { + snapshot := o.Snapshot(ctx, plugin.ID, handlers, builds, gc) + operations, _ := o.repo.ListOperations(ctx, plugin.ID, 50) + body := map[string]any{ + "created_at": time.Now().Unix(), + "plugin": plugin, + "manifest": manifest, + "operations": snapshot, + "recent_operations": operations, + "redaction_policy": []string{ + "secret", "token", "password", "session response", "full packet payload", + }, + } + data, err := json.MarshalIndent(body, "", " ") + if err != nil { + return nil, DiagnosticPackageSummary{}, err + } + data = []byte(redactSensitive(string(data))) + dir := filepath.Join(o.runtimeRoot(), "diagnostics", plugin.ID) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, DiagnosticPackageSummary{}, err + } + file := filepath.Join(dir, fmt.Sprintf("%d.json", time.Now().UnixNano())) + if err := os.WriteFile(file, data, 0600); err != nil { + return nil, DiagnosticPackageSummary{}, err + } + summary, err := o.repo.SaveDiagnostic(ctx, plugin.ID, file, int64(len(data)), []string{"plugin", "manifest", "operations", "recent_operations"}) + if err != nil { + return nil, DiagnosticPackageSummary{}, err + } + return data, summary, nil +} + +func (o *Operations) runtimeRoot() string { + if o.root == "" { + return os.TempDir() + } + return filepath.Join(o.root, "_runtime") +} + +func (o *Operations) GCCandidates(ctx context.Context, pluginID string) ([]GCCandidate, error) { + now := time.Now().Unix() + var candidates []GCCandidate + data, err := o.repo.ListPluginData(ctx, pluginID) + if err != nil { + return nil, err + } + for _, record := range data { + expired := record.ExpiresAt > 0 && record.ExpiresAt <= now + reason := "plugin_data retained" + if expired { + reason = "plugin_data retention expired" + } + candidates = append(candidates, GCCandidate{ + Kind: "plugin_data", + ID: record.Key, + PluginID: record.PluginID, + Protected: !expired, + Reason: reason, + SizeBytes: record.SizeBytes, + CreatedAt: record.UpdatedAt, + Referenced: !expired, + }) + } + files, err := o.repo.ListPluginFiles(ctx, pluginID) + if err != nil { + return nil, err + } + seenFiles := make(map[string]bool) + for _, record := range files { + expired := record.ExpiresAt > 0 && record.ExpiresAt <= now + filePath := filepath.Join(o.runtimeRoot(), record.PluginID, record.Namespace, filepath.FromSlash(record.Path)) + seenFiles[filePath] = true + reason := "plugin file retained" + if expired { + reason = "plugin file retention expired" + } + candidates = append(candidates, GCCandidate{ + Kind: "plugin_file", + ID: record.Namespace + "/" + record.Path, + PluginID: record.PluginID, + Path: filePath, + Protected: !expired, + Reason: reason, + SizeBytes: record.SizeBytes, + CreatedAt: record.UpdatedAt, + Referenced: !expired, + }) + } + runtimeRoot := o.runtimeRoot() + _ = filepath.WalkDir(runtimeRoot, func(filePath string, d os.DirEntry, err error) error { + if err != nil || d.IsDir() || !strings.Contains(filePath, string(filepath.Separator)+"_runtime"+string(filepath.Separator)) { + return nil + } + if seenFiles[filePath] { + return nil + } + info, err := d.Info() + if err != nil { + return nil + } + rel, _ := filepath.Rel(runtimeRoot, filePath) + parts := strings.Split(rel, string(filepath.Separator)) + id := rel + pid := "" + if len(parts) > 0 { + pid = parts[0] + } + if pluginID != "" && pid != pluginID { + return nil + } + candidates = append(candidates, GCCandidate{ + Kind: "plugin_file_orphan", + ID: id, + PluginID: pid, + Path: filePath, + Protected: false, + Reason: "orphaned runtime file", + SizeBytes: info.Size(), + CreatedAt: info.ModTime().Unix(), + }) + return nil + }) + logs, err := o.repo.RecentLogs(ctx, pluginID, DefaultLogRecentLimit+1) + if err == nil && len(logs) > DefaultLogRecentLimit { + for _, item := range logs[DefaultLogRecentLimit:] { + candidates = append(candidates, GCCandidate{ + Kind: "plugin_log", + ID: fmt.Sprintf("%s/%d", item.PluginID, item.CreatedAt), + PluginID: item.PluginID, + Protected: false, + Reason: "log summary exceeds recent retention", + SizeBytes: int64(len(item.Message)), + CreatedAt: item.CreatedAt, + }) + } + } + return candidates, nil +} + +func (o *Operations) RunGC(ctx context.Context, actor, pluginID string, dryRun bool) ([]GCCandidate, error) { + candidates, err := o.GCCandidates(ctx, pluginID) + if err != nil { + return nil, err + } + if dryRun { + _ = o.repo.RecordOperation(ctx, pluginID, "", "plugin_operations_gc", "dry_run", actor, "plugin operations gc dry-run completed", map[string]any{"candidates": len(candidates)}) + return candidates, nil + } + var removed []GCCandidate + for _, candidate := range candidates { + if candidate.Protected { + continue + } + switch candidate.Kind { + case "plugin_data": + _ = o.repo.DeletePluginData(ctx, candidate.PluginID, candidate.ID) + removed = append(removed, candidate) + case "plugin_file": + namespace, name, _ := strings.Cut(candidate.ID, "/") + _ = os.Remove(candidate.Path) + _ = o.repo.DeletePluginFile(ctx, candidate.PluginID, namespace, name) + removed = append(removed, candidate) + case "plugin_file_orphan": + _ = os.Remove(candidate.Path) + removed = append(removed, candidate) + } + } + _ = o.repo.RecordOperation(ctx, pluginID, "", "plugin_operations_gc", "succeeded", actor, "plugin operations gc completed", map[string]any{"removed": len(removed)}) + return removed, nil +} + +func (po *PluginOperations) validateEvent(name string, fields map[string]string) (map[string]string, string, error) { + if !metricNamePattern.MatchString(name) { + return nil, "invalid_event_name", fmt.Errorf("invalid event name %q", name) + } + po.mu.Lock() + defer po.mu.Unlock() + allowed, declared := po.eventSchemas[name] + if !declared { + return nil, "undeclared_event", fmt.Errorf("event %q is not declared by manifest", name) + } + clean, err := sanitizeLabels(fields, allowed) + if err != nil { + return clean, "invalid_fields", err + } + values := po.eventValues[name] + if values == nil { + values = make(map[string]map[string]struct{}) + po.eventValues[name] = values + } + for field, value := range clean { + set := values[field] + if set == nil { + set = make(map[string]struct{}) + values[field] = set + } + set[value] = struct{}{} + if len(set) > 64 { + return clean, "high_cardinality_field", fmt.Errorf("event %q field %q exceeded low-cardinality limit", name, field) + } + } + return clean, "", nil +} + +func (po *PluginOperations) validateMetric(name string, labels map[string]string) (map[string]string, string, error) { + if !metricNamePattern.MatchString(name) { + return nil, "", fmt.Errorf("invalid metric name %q", name) + } + po.mu.Lock() + defer po.mu.Unlock() + spec, declared := po.metricSchemas[name] + if !declared { + return nil, "", fmt.Errorf("custom metric %q is not declared by manifest", name) + } + allowed := make(map[string]bool, len(spec.Labels)) + for _, label := range spec.Labels { + allowed[label] = true + } + clean, err := sanitizeLabels(labels, allowed) + if err != nil { + return clean, spec.Type, err + } + return clean, spec.Type, nil +} + +type pluginLogger struct { + ops *PluginOperations +} + +func (l pluginLogger) Debug(ctx context.Context, message string, fields map[string]string) { + l.write(ctx, "debug", message, fields) +} + +func (l pluginLogger) Info(ctx context.Context, message string, fields map[string]string) { + l.write(ctx, "info", message, fields) +} + +func (l pluginLogger) Warn(ctx context.Context, message string, fields map[string]string) { + l.write(ctx, "warn", message, fields) +} + +func (l pluginLogger) Error(ctx context.Context, message string, fields map[string]string) { + l.write(ctx, "error", message, fields) +} + +func (l pluginLogger) write(ctx context.Context, level, message string, fields map[string]string) { + if l.ops == nil { + return + } + trace := traceFromContext(ctx) + clean, _ := sanitizeLabels(fields, nil) + item := LogSummary{ + PluginID: l.ops.pluginID, + Level: level, + Message: redactSensitive(limitString(message, 512)), + Fields: clean, + TraceID: trace.TraceID, + ConnectionID: trace.ConnectionID, + CreatedAt: time.Now().Unix(), + } + _ = l.ops.parent.repo.SaveLog(context.Background(), item) + log.Info(). + Str("plugin", l.ops.pluginID). + Str("level", level). + Str("trace_id", trace.TraceID). + Str("connection_id", trace.ConnectionID). + Msg(item.Message) +} + +type pluginDataStore struct { + ops *PluginOperations +} + +func (s pluginDataStore) Put(ctx context.Context, record api.DataRecord) error { + if s.ops == nil { + return errors.New("plugin data store is unavailable") + } + key, err := cleanStoreKey(record.Key) + if err != nil { + return err + } + if len(record.Value) > DefaultPluginDataKeyLimit { + return fmt.Errorf("plugin_data key %q size %d exceeds limit %d", key, len(record.Value), DefaultPluginDataKeyLimit) + } + current, _ := s.ops.parent.repo.PluginDataUsage(ctx, s.ops.pluginID) + old, _, _ := s.ops.parent.repo.GetPluginData(ctx, s.ops.pluginID, key) + nextUsage := current - old.SizeBytes + int64(len(record.Value)) + if nextUsage > s.ops.dataQuota { + return fmt.Errorf("plugin_data quota exceeded: %d > %d", nextUsage, s.ops.dataQuota) + } + expiresAt := retentionDeadline(record.Retention) + if record.DataClass == "" { + record.DataClass = s.ops.dataSpec.DataClass + } + if record.SchemaVersion == 0 { + record.SchemaVersion = s.ops.dataSpec.SchemaVersion + } + return s.ops.parent.repo.PutPluginData(ctx, PluginDataSummary{ + PluginID: s.ops.pluginID, + Key: key, + SchemaVersion: record.SchemaVersion, + DataClass: redactSensitive(record.DataClass), + Exportable: record.Exportable || s.ops.dataSpec.Exportable, + SizeBytes: int64(len(record.Value)), + ExpiresAt: expiresAt, + }, append([]byte(nil), record.Value...)) +} + +func (s pluginDataStore) Get(ctx context.Context, key string) (api.DataRecord, error) { + if s.ops == nil { + return api.DataRecord{}, errors.New("plugin data store is unavailable") + } + clean, err := cleanStoreKey(key) + if err != nil { + return api.DataRecord{}, err + } + record, value, err := s.ops.parent.repo.GetPluginData(ctx, s.ops.pluginID, clean) + if err != nil { + return api.DataRecord{}, err + } + return api.DataRecord{ + Key: record.Key, + Value: append([]byte(nil), value...), + SchemaVersion: record.SchemaVersion, + DataClass: record.DataClass, + Exportable: record.Exportable, + }, nil +} + +func (s pluginDataStore) Delete(ctx context.Context, key string) error { + if s.ops == nil { + return errors.New("plugin data store is unavailable") + } + clean, err := cleanStoreKey(key) + if err != nil { + return err + } + return s.ops.parent.repo.DeletePluginData(ctx, s.ops.pluginID, clean) +} + +type pluginFileStore struct { + ops *PluginOperations +} + +func (s pluginFileStore) ResourcePath(name string) (string, error) { + if s.ops == nil { + return "", errors.New("plugin file store is unavailable") + } + clean, err := cleanStorePath(name) + if err != nil { + return "", err + } + artifactDir := filepath.Join(s.ops.parent.root, s.ops.pluginID, s.ops.artifactID) + resource := filepath.Join(artifactDir, "resources", filepath.FromSlash(clean)) + if !isSubpath(filepath.Join(artifactDir, "resources"), resource) { + return "", errors.New("unsafe resource path") + } + return resource, nil +} + +func (s pluginFileStore) Write(ctx context.Context, namespace, name string, data []byte, dataClass string, retention time.Duration) error { + if s.ops == nil { + return errors.New("plugin file store is unavailable") + } + namespace, spec, err := s.namespaceSpec(namespace) + if err != nil { + return err + } + if spec.Readonly { + return fmt.Errorf("file namespace %q is readonly", namespace) + } + clean, err := cleanStorePath(name) + if err != nil { + return err + } + quota := spec.QuotaBytes + if quota <= 0 { + quota = s.ops.fileQuota + } + usage, _ := s.ops.parent.repo.PluginFileUsage(ctx, s.ops.pluginID) + if usage+int64(len(data)) > quota { + return fmt.Errorf("plugin file quota exceeded: %d > %d", usage+int64(len(data)), quota) + } + root := s.runtimeRoot(namespace) + target := filepath.Join(root, filepath.FromSlash(clean)) + if !isSubpath(root, target) { + return errors.New("unsafe file path") + } + if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil { + return err + } + if err := os.WriteFile(target, data, 0644); err != nil { + return err + } + if dataClass == "" { + dataClass = spec.DataClass + } + return s.ops.parent.repo.UpsertPluginFile(ctx, PluginFileSummary{ + PluginID: s.ops.pluginID, + Namespace: namespace, + Path: clean, + DataClass: redactSensitive(dataClass), + SizeBytes: int64(len(data)), + ExpiresAt: retentionDeadline(retention), + Readonly: spec.Readonly, + }, target) +} + +func (s pluginFileStore) Read(ctx context.Context, namespace, name string, maxBytes int64) ([]byte, error) { + _ = ctx + if s.ops == nil { + return nil, errors.New("plugin file store is unavailable") + } + namespace, _, err := s.namespaceSpec(namespace) + if err != nil { + return nil, err + } + clean, err := cleanStorePath(name) + if err != nil { + return nil, err + } + root := s.runtimeRoot(namespace) + target := filepath.Join(root, filepath.FromSlash(clean)) + if !isSubpath(root, target) { + return nil, errors.New("unsafe file path") + } + if maxBytes <= 0 { + maxBytes = DefaultPluginDataKeyLimit + } + file, err := os.Open(target) + if err != nil { + return nil, err + } + defer file.Close() + var buf bytes.Buffer + if _, err := io.CopyN(&buf, file, maxBytes+1); err != nil && !errors.Is(err, io.EOF) { + return nil, err + } + if int64(buf.Len()) > maxBytes { + return nil, fmt.Errorf("file %q exceeds max read size %d", clean, maxBytes) + } + return buf.Bytes(), nil +} + +func (s pluginFileStore) Delete(ctx context.Context, namespace, name string) error { + if s.ops == nil { + return errors.New("plugin file store is unavailable") + } + namespace, spec, err := s.namespaceSpec(namespace) + if err != nil { + return err + } + if spec.Readonly { + return fmt.Errorf("file namespace %q is readonly", namespace) + } + clean, err := cleanStorePath(name) + if err != nil { + return err + } + root := s.runtimeRoot(namespace) + target := filepath.Join(root, filepath.FromSlash(clean)) + if !isSubpath(root, target) { + return errors.New("unsafe file path") + } + _ = os.Remove(target) + return s.ops.parent.repo.DeletePluginFile(ctx, s.ops.pluginID, namespace, clean) +} + +func (s pluginFileStore) namespaceSpec(namespace string) (string, FileStoreSpec, error) { + namespace = strings.TrimSpace(namespace) + if namespace == "" { + namespace = "data" + } + if !metricNamePattern.MatchString(namespace) { + return "", FileStoreSpec{}, fmt.Errorf("invalid file namespace %q", namespace) + } + spec := s.ops.fileSpecs[namespace] + if spec.Namespace == "" { + return "", FileStoreSpec{}, fmt.Errorf("file namespace %q is not declared", namespace) + } + return namespace, spec, nil +} + +func (s pluginFileStore) runtimeRoot(namespace string) string { + root := s.ops.parent.root + if root == "" { + root = os.TempDir() + } + return filepath.Join(root, "_runtime", s.ops.pluginID, namespace) +} + +type pluginExternalClient struct { + ops *PluginOperations + name string + runtime *externalRuntime +} + +func (c pluginExternalClient) DoHTTP(ctx context.Context, req api.ExternalRequest) (api.ExternalResponse, error) { + if c.ops == nil || c.runtime == nil { + return api.ExternalResponse{}, errors.New("external client is unavailable") + } + spec, err := c.declaredSpec() + if err != nil { + return api.ExternalResponse{}, err + } + if req.URL == "" { + req.URL = spec.Endpoint + } + if req.Method == "" { + req.Method = http.MethodGet + } + if err := c.beforeRequest(); err != nil { + return api.ExternalResponse{}, err + } + start := time.Now() + defer c.runtime.inflight.Add(-1) + + timeout := req.Timeout + if timeout <= 0 { + timeout = parseDurationDefault(spec.Timeout, DefaultHandlerTimeout) + } + if timeout <= 0 { + timeout = DefaultHandlerTimeout + } + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + httpReq, err := http.NewRequestWithContext(ctx, req.Method, req.URL, req.Body) + if err != nil { + c.finish(start, "request_error", err) + return api.ExternalResponse{}, err + } + httpReq.Header = req.Header.Clone() + if spec.Traceparent { + trace := traceFromContext(ctx) + if trace.TraceID != "" { + httpReq.Header.Set("traceparent", "00-"+limitHex(trace.TraceID, 32)+"-0000000000000000-01") + } + } + + attempts := spec.Retry + 1 + if attempts <= 0 { + attempts = 1 + } + var lastErr error + var resp *http.Response + for i := 0; i < attempts; i++ { + resp, lastErr = http.DefaultClient.Do(httpReq) + if lastErr == nil && resp.StatusCode < 500 { + break + } + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + } + if lastErr != nil { + c.finish(start, "error", lastErr) + return api.ExternalResponse{}, lastErr + } + defer resp.Body.Close() + body, err := readLimited(resp.Body, 1024*1024) + if err != nil { + c.finish(start, "read_error", err) + return api.ExternalResponse{}, err + } + if resp.StatusCode >= 500 { + err := fmt.Errorf("external dependency %s returned status %d", c.name, resp.StatusCode) + c.finish(start, fmt.Sprintf("http_%d", resp.StatusCode), err) + return api.ExternalResponse{}, err + } + c.finish(start, fmt.Sprintf("http_%d", resp.StatusCode), nil) + c.recordTrace(ctx, start, "http", fmt.Sprintf("%d", resp.StatusCode)) + return api.ExternalResponse{StatusCode: resp.StatusCode, Header: resp.Header.Clone(), Body: body}, nil +} + +func (c pluginExternalClient) DialTCP(ctx context.Context, address string, timeout time.Duration) (net.Conn, error) { + if c.ops == nil || c.runtime == nil { + return nil, errors.New("external client is unavailable") + } + spec, err := c.declaredSpec() + if err != nil { + return nil, err + } + if address == "" { + address = strings.TrimPrefix(spec.Endpoint, "tcp://") + } + if timeout <= 0 { + timeout = parseDurationDefault(spec.Timeout, DefaultExternalTimeout) + } + if err := c.beforeRequest(); err != nil { + return nil, err + } + start := time.Now() + defer c.runtime.inflight.Add(-1) + var dialer net.Dialer + conn, err := dialer.DialContext(ctx, "tcp", address) + if err != nil { + c.finish(start, "dial_error", err) + return nil, err + } + c.finish(start, "ok", nil) + c.recordTrace(ctx, start, "tcp", "ok") + return conn, nil +} + +func (c pluginExternalClient) HealthCheck(ctx context.Context) error { + spec, err := c.declaredSpec() + if err != nil { + return err + } + if strings.HasPrefix(spec.Endpoint, "tcp://") { + conn, err := c.DialTCP(ctx, strings.TrimPrefix(spec.Endpoint, "tcp://"), parseDurationDefault(spec.Timeout, DefaultExternalTimeout)) + if conn != nil { + _ = conn.Close() + } + return err + } + resp, err := c.DoHTTP(ctx, api.ExternalRequest{Method: http.MethodHead, URL: spec.Endpoint}) + if err == nil && resp.StatusCode >= 400 { + err = fmt.Errorf("health check status %d", resp.StatusCode) + } + if err != nil { + c.runtime.mu.Lock() + c.runtime.lastStatus = "health_failed" + c.runtime.recentError = redactSensitive(err.Error()) + c.runtime.mu.Unlock() + } + return err +} + +func (c pluginExternalClient) declaredSpec() (ExternalSpec, error) { + c.ops.mu.Lock() + defer c.ops.mu.Unlock() + spec := c.ops.externalSpecs[c.name] + if spec.Name == "" { + return ExternalSpec{}, fmt.Errorf("external dependency %q is not declared by manifest", c.name) + } + return spec, nil +} + +func (c pluginExternalClient) beforeRequest() error { + c.runtime.mu.Lock() + defer c.runtime.mu.Unlock() + if time.Now().Before(c.runtime.circuitUntil) { + return fmt.Errorf("external dependency %s circuit is open", c.name) + } + c.runtime.requests.Add(1) + c.runtime.inflight.Add(1) + return nil +} + +func (c pluginExternalClient) finish(start time.Time, status string, err error) { + duration := time.Since(start) + c.runtime.durationCount.Add(1) + c.runtime.durationSumMS.Add(uint64(duration.Milliseconds())) + c.runtime.mu.Lock() + defer c.runtime.mu.Unlock() + c.runtime.lastStatus = status + c.runtime.lastSeenAt = time.Now().Unix() + if err != nil { + c.runtime.errors.Add(1) + failures := c.runtime.consecutiveFailures.Add(1) + c.runtime.recentError = redactSensitive(err.Error()) + if failures >= 3 { + c.runtime.circuitUntil = time.Now().Add(30 * time.Second) + } + return + } + c.runtime.consecutiveFailures.Store(0) + c.runtime.recentError = "" + c.runtime.circuitUntil = time.Time{} +} + +func (c pluginExternalClient) recordTrace(ctx context.Context, start time.Time, kind, status string) { + trace := traceFromContext(ctx) + _ = c.ops.parent.repo.SaveTrace(context.Background(), TraceSummary{ + PluginID: c.ops.pluginID, + TraceID: trace.TraceID, + ConnectionID: trace.ConnectionID, + HandlerID: trace.HandlerID, + Operation: "external." + kind + "." + c.name, + Status: status, + DurationMS: time.Since(start).Milliseconds(), + }, map[string]string{ + "endpoint": redactEndpoint(c.runtime.spec.Endpoint), + "purpose": redactSensitive(c.runtime.spec.Purpose), + }) +} + +func (rt *externalRuntime) summary(pluginID, name string) ExternalDependencySummary { + rt.mu.Lock() + defer rt.mu.Unlock() + state := circuitClosed + if time.Now().Before(rt.circuitUntil) { + state = circuitOpen + } + return ExternalDependencySummary{ + PluginID: pluginID, + Name: name, + Endpoint: redactEndpoint(rt.spec.Endpoint), + Purpose: redactSensitive(rt.spec.Purpose), + Required: rt.spec.Required, + FailPolicy: rt.spec.FailPolicy, + DataClasses: append([]string(nil), rt.spec.DataClasses...), + Requests: rt.requests.Load(), + Errors: rt.errors.Load(), + Inflight: rt.inflight.Load(), + DurationCount: rt.durationCount.Load(), + DurationSumMS: rt.durationSumMS.Load(), + CircuitState: state, + ConsecutiveFailures: rt.consecutiveFailures.Load(), + RecentError: rt.recentError, + LastStatus: rt.lastStatus, + LastSeenAt: rt.lastSeenAt, + } +} + +func (rt *taskRuntime) summary() BackgroundTaskSummary { + rt.mu.Lock() + defer rt.mu.Unlock() + mode := rt.spec.Mode + if mode == "" { + if rt.task.Manual { + mode = "manual" + } else { + mode = "interval" + } + } + return BackgroundTaskSummary{ + PluginID: rt.pluginID, + TaskID: rt.task.ID, + Name: rt.task.Name, + Mode: mode, + IntervalMS: rt.task.Interval.Milliseconds(), + RunOnStart: rt.task.RunOnStart, + TimeoutMS: rt.task.Timeout.Milliseconds(), + Manual: rt.task.Manual, + Running: rt.running, + LastRunAt: rt.lastRunAt, + NextRunAt: rt.nextRunAt, + LastDurationMS: rt.lastDurationMS, + LastError: rt.lastError, + Skipped: rt.skipped, + ConsecutiveFailures: rt.consecutiveFailures, + } +} + +func WithTraceContext(ctx context.Context, pluginID, traceID, connectionID, handlerID string) context.Context { + return context.WithValue(ctx, traceContextKey{}, traceContext{ + PluginID: pluginID, + TraceID: traceID, + ConnectionID: connectionID, + HandlerID: handlerID, + }) +} + +func traceFromContext(ctx context.Context) traceContext { + if ctx == nil { + return traceContext{} + } + trace, _ := ctx.Value(traceContextKey{}).(traceContext) + return trace +} + +func sanitizeLabels(fields map[string]string, allowed map[string]bool) (map[string]string, error) { + if len(fields) == 0 { + return map[string]string{}, nil + } + if len(fields) > 12 { + return nil, errors.New("too many fields") + } + clean := make(map[string]string, len(fields)) + for key, value := range fields { + if !metricNamePattern.MatchString(key) { + return clean, fmt.Errorf("invalid field %q", key) + } + if len(allowed) > 0 && !allowed[key] { + return clean, fmt.Errorf("field %q is not declared", key) + } + if isSensitiveName(key) { + return clean, fmt.Errorf("field %q is sensitive", key) + } + if len(value) > DefaultLabelValueMaxBytes { + return clean, fmt.Errorf("field %q value exceeds low-cardinality size limit", key) + } + value = redactSensitive(value) + if value == "" { + continue + } + clean[key] = value + } + return clean, nil +} + +func cleanStoreKey(key string) (string, error) { + key = strings.TrimSpace(key) + if key == "" || !storeKeyPattern.MatchString(key) { + return "", fmt.Errorf("invalid plugin_data key %q", key) + } + if strings.Contains(key, "..") || strings.HasPrefix(key, "/") || strings.Contains(key, `\`) { + return "", fmt.Errorf("unsafe plugin_data key %q", key) + } + return key, nil +} + +func cleanStorePath(name string) (string, error) { + if name == "" || strings.Contains(name, `\`) || strings.HasPrefix(name, "/") { + return "", fmt.Errorf("unsafe file path %q", name) + } + clean := path.Clean(name) + if clean == "." || clean != name || strings.HasPrefix(clean, "../") || clean == ".." || path.IsAbs(clean) { + return "", fmt.Errorf("unsafe file path %q", name) + } + return clean, nil +} + +func isSubpath(root, target string) bool { + root = filepath.Clean(root) + target = filepath.Clean(target) + rel, err := filepath.Rel(root, target) + return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) +} + +func retentionDeadline(retention time.Duration) int64 { + if retention <= 0 { + return 0 + } + return time.Now().Add(retention).Unix() +} + +func parseDurationDefault(value string, fallback time.Duration) time.Duration { + if value == "" { + return fallback + } + parsed, err := time.ParseDuration(value) + if err != nil { + return fallback + } + return parsed +} + +func deterministicJitter(jitter time.Duration, key string) time.Duration { + if jitter <= 0 || key == "" { + return 0 + } + var sum int64 + for _, ch := range key { + sum += int64(ch) + } + return time.Duration(sum % int64(jitter)) +} + +func limitString(value string, max int) string { + if max <= 0 || len(value) <= max { + return value + } + return value[:max] +} + +func redactSensitive(value string) string { + if value == "" { + return "" + } + lower := strings.ToLower(value) + for _, marker := range []string{"secret", "token", "password", "session", "credential", "authorization", "packet"} { + if strings.Contains(lower, marker) { + return "[REDACTED]" + } + } + return value +} + +func redactEndpoint(endpoint string) string { + if endpoint == "" { + return "" + } + if strings.Contains(endpoint, "@") { + return "[REDACTED]" + } + return limitString(endpoint, 256) +} + +func randomToken() string { + var data [16]byte + if _, err := rand.Read(data[:]); err != nil { + return fmt.Sprintf("%d", time.Now().UnixNano()) + } + return hex.EncodeToString(data[:]) +} + +func limitHex(value string, max int) string { + value = strings.ToLower(value) + var out strings.Builder + for _, ch := range value { + if (ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f') { + out.WriteRune(ch) + } + } + for out.Len() < max { + out.WriteByte('0') + } + return out.String()[:max] +} + +func readLimited(reader io.Reader, max int64) ([]byte, error) { + var buf bytes.Buffer + if _, err := io.CopyN(&buf, reader, max+1); err != nil && !errors.Is(err, io.EOF) { + return nil, err + } + if int64(buf.Len()) > max { + return nil, fmt.Errorf("response exceeds limit %d", max) + } + return buf.Bytes(), nil +} + +func mergeEventSummaries(current, recent []EventSummary) []EventSummary { + byKey := make(map[string]EventSummary) + for _, event := range current { + byKey[event.PluginID+"\x00"+event.Name] = event + } + for _, event := range recent { + key := event.PluginID + "\x00" + event.Name + existing := byKey[key] + existing.PluginID = event.PluginID + existing.Name = event.Name + existing.Count += event.Count + existing.Dropped += event.Dropped + if event.LastSeenAt > existing.LastSeenAt { + existing.LastSeenAt = event.LastSeenAt + existing.Fields = event.Fields + } + byKey[key] = existing + } + out := make([]EventSummary, 0, len(byKey)) + for _, event := range byKey { + out = append(out, event) + } + return out +} + +type noopOperationsLogger struct{} + +func (noopOperationsLogger) Debug(context.Context, string, map[string]string) {} +func (noopOperationsLogger) Info(context.Context, string, map[string]string) {} +func (noopOperationsLogger) Warn(context.Context, string, map[string]string) {} +func (noopOperationsLogger) Error(context.Context, string, map[string]string) {} + +type noopOperationsDataStore struct{} + +func (noopOperationsDataStore) Put(context.Context, api.DataRecord) error { return nil } +func (noopOperationsDataStore) Get(context.Context, string) (api.DataRecord, error) { + return api.DataRecord{}, errors.New("plugin data store is unavailable") +} +func (noopOperationsDataStore) Delete(context.Context, string) error { return nil } + +type noopOperationsFileStore struct{} + +func (noopOperationsFileStore) ResourcePath(string) (string, error) { + return "", errors.New("plugin file store is unavailable") +} +func (noopOperationsFileStore) Write(context.Context, string, string, []byte, string, time.Duration) error { + return nil +} +func (noopOperationsFileStore) Read(context.Context, string, string, int64) ([]byte, error) { + return nil, errors.New("plugin file store is unavailable") +} +func (noopOperationsFileStore) Delete(context.Context, string, string) error { return nil } + +type noopOperationsExternalClient struct{} + +func (noopOperationsExternalClient) DoHTTP(context.Context, api.ExternalRequest) (api.ExternalResponse, error) { + return api.ExternalResponse{}, errors.New("external client is unavailable") +} +func (noopOperationsExternalClient) DialTCP(context.Context, string, time.Duration) (net.Conn, error) { + return nil, errors.New("external client is unavailable") +} +func (noopOperationsExternalClient) HealthCheck(context.Context) error { return nil } diff --git a/internal/pluginmanager/repository.go b/internal/pluginmanager/repository.go index 47ab4c7..3e6bb5d 100644 --- a/internal/pluginmanager/repository.go +++ b/internal/pluginmanager/repository.go @@ -965,6 +965,339 @@ VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, return err } +func (r Repository) ListOperations(ctx context.Context, pluginID string, limit int) ([]OperationRecord, error) { + if limit <= 0 { + limit = 50 + } + query := ` +SELECT id, plugin_id, artifact_id, operation, status, actor, message, metadata_json, created_at +FROM plugin_operations` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY created_at DESC, id DESC LIMIT ?` + args = append(args, limit) + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var records []OperationRecord + for rows.Next() { + var record OperationRecord + if err := rows.Scan(&record.ID, &record.PluginID, &record.ArtifactID, &record.Operation, &record.Status, &record.Actor, &record.Message, &record.MetadataJSON, &record.CreatedAt); err != nil { + return nil, err + } + records = append(records, record) + } + return records, rows.Err() +} + +func (r Repository) SaveEvent(ctx context.Context, event EventSummary, dropped bool, reason, traceID, connectionID string) error { + fields, err := marshalDefaultObject(event.Fields) + if err != nil { + return err + } + _, err = r.db.ExecContext(ctx, ` +INSERT INTO plugin_events(plugin_id, name, fields_json, dropped, reason, trace_id, connection_id, created_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + event.PluginID, event.Name, fields, boolInt(dropped), reason, traceID, connectionID, r.now().Unix()) + return err +} + +func (r Repository) RecentEvents(ctx context.Context, pluginID string, limit int) ([]EventSummary, error) { + if limit <= 0 { + limit = DefaultEventRecentLimit + } + query := `SELECT plugin_id, name, fields_json, dropped, reason, created_at FROM plugin_events` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY created_at DESC, id DESC LIMIT ?` + args = append(args, limit) + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var events []EventSummary + for rows.Next() { + var event EventSummary + var fieldsJSON string + var dropped int + var reason string + if err := rows.Scan(&event.PluginID, &event.Name, &fieldsJSON, &dropped, &reason, &event.LastSeenAt); err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(defaultJSONObject(fieldsJSON)), &event.Fields) + if dropped != 0 { + event.Dropped = 1 + if reason != "" { + if event.Fields == nil { + event.Fields = make(map[string]string) + } + event.Fields["drop_reason"] = reason + } + } else { + event.Count = 1 + } + events = append(events, event) + } + return events, rows.Err() +} + +func (r Repository) SaveLog(ctx context.Context, log LogSummary) error { + fields, err := marshalDefaultObject(log.Fields) + if err != nil { + return err + } + _, err = r.db.ExecContext(ctx, ` +INSERT INTO plugin_logs(plugin_id, level, message, fields_json, trace_id, connection_id, created_at) +VALUES (?, ?, ?, ?, ?, ?, ?)`, + log.PluginID, log.Level, log.Message, fields, log.TraceID, log.ConnectionID, r.now().Unix()) + return err +} + +func (r Repository) RecentLogs(ctx context.Context, pluginID string, limit int) ([]LogSummary, error) { + if limit <= 0 { + limit = DefaultLogRecentLimit + } + query := `SELECT plugin_id, level, message, fields_json, trace_id, connection_id, created_at FROM plugin_logs` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY created_at DESC, id DESC LIMIT ?` + args = append(args, limit) + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var logs []LogSummary + for rows.Next() { + var item LogSummary + var fieldsJSON string + if err := rows.Scan(&item.PluginID, &item.Level, &item.Message, &fieldsJSON, &item.TraceID, &item.ConnectionID, &item.CreatedAt); err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(defaultJSONObject(fieldsJSON)), &item.Fields) + logs = append(logs, item) + } + return logs, rows.Err() +} + +func (r Repository) SaveTrace(ctx context.Context, trace TraceSummary, fields map[string]string) error { + fieldsJSON, err := marshalDefaultObject(fields) + if err != nil { + return err + } + _, err = r.db.ExecContext(ctx, ` +INSERT INTO plugin_traces(plugin_id, trace_id, connection_id, handler_id, operation, status, duration_ms, fields_json, created_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + trace.PluginID, trace.TraceID, trace.ConnectionID, trace.HandlerID, trace.Operation, trace.Status, trace.DurationMS, fieldsJSON, r.now().Unix()) + return err +} + +func (r Repository) RecentTraces(ctx context.Context, pluginID string, limit int) ([]TraceSummary, error) { + if limit <= 0 { + limit = 200 + } + query := `SELECT plugin_id, trace_id, connection_id, handler_id, operation, status, duration_ms, created_at FROM plugin_traces` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY created_at DESC, id DESC LIMIT ?` + args = append(args, limit) + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var traces []TraceSummary + for rows.Next() { + var trace TraceSummary + if err := rows.Scan(&trace.PluginID, &trace.TraceID, &trace.ConnectionID, &trace.HandlerID, &trace.Operation, &trace.Status, &trace.DurationMS, &trace.CreatedAt); err != nil { + return nil, err + } + traces = append(traces, trace) + } + return traces, rows.Err() +} + +func (r Repository) PutPluginData(ctx context.Context, record PluginDataSummary, value []byte) error { + now := r.now().Unix() + _, err := r.db.ExecContext(ctx, ` +INSERT INTO plugin_data(plugin_id, key, value, schema_version, data_class, exportable, size_bytes, expires_at, created_at, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(plugin_id, key) DO UPDATE SET + value = excluded.value, + schema_version = excluded.schema_version, + data_class = excluded.data_class, + exportable = excluded.exportable, + size_bytes = excluded.size_bytes, + expires_at = excluded.expires_at, + updated_at = excluded.updated_at`, + record.PluginID, record.Key, value, record.SchemaVersion, record.DataClass, boolInt(record.Exportable), int64(len(value)), record.ExpiresAt, now, now) + return err +} + +func (r Repository) GetPluginData(ctx context.Context, pluginID, key string) (PluginDataSummary, []byte, error) { + row := r.db.QueryRowContext(ctx, ` +SELECT plugin_id, key, value, schema_version, data_class, exportable, size_bytes, expires_at, updated_at +FROM plugin_data WHERE plugin_id = ? AND key = ?`, pluginID, key) + var record PluginDataSummary + var exportable int + var value []byte + err := row.Scan(&record.PluginID, &record.Key, &value, &record.SchemaVersion, &record.DataClass, &exportable, &record.SizeBytes, &record.ExpiresAt, &record.UpdatedAt) + record.Exportable = exportable != 0 + return record, value, err +} + +func (r Repository) DeletePluginData(ctx context.Context, pluginID, key string) error { + _, err := r.db.ExecContext(ctx, `DELETE FROM plugin_data WHERE plugin_id = ? AND key = ?`, pluginID, key) + return err +} + +func (r Repository) ListPluginData(ctx context.Context, pluginID string) ([]PluginDataSummary, error) { + query := `SELECT plugin_id, key, schema_version, data_class, exportable, size_bytes, expires_at, updated_at FROM plugin_data` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY updated_at DESC, key ASC` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var records []PluginDataSummary + for rows.Next() { + var record PluginDataSummary + var exportable int + if err := rows.Scan(&record.PluginID, &record.Key, &record.SchemaVersion, &record.DataClass, &exportable, &record.SizeBytes, &record.ExpiresAt, &record.UpdatedAt); err != nil { + return nil, err + } + record.Exportable = exportable != 0 + records = append(records, record) + } + return records, rows.Err() +} + +func (r Repository) PluginDataUsage(ctx context.Context, pluginID string) (int64, error) { + row := r.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(size_bytes), 0) FROM plugin_data WHERE plugin_id = ?`, pluginID) + var total int64 + return total, row.Scan(&total) +} + +func (r Repository) UpsertPluginFile(ctx context.Context, record PluginFileSummary, diskPath string) error { + now := r.now().Unix() + _, err := r.db.ExecContext(ctx, ` +INSERT INTO plugin_files(plugin_id, namespace, path, disk_path, data_class, exportable, readonly, size_bytes, expires_at, created_at, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(plugin_id, namespace, path) DO UPDATE SET + disk_path = excluded.disk_path, + data_class = excluded.data_class, + exportable = excluded.exportable, + readonly = excluded.readonly, + size_bytes = excluded.size_bytes, + expires_at = excluded.expires_at, + updated_at = excluded.updated_at`, + record.PluginID, record.Namespace, record.Path, diskPath, record.DataClass, boolInt(record.Readonly), boolInt(record.Readonly), record.SizeBytes, record.ExpiresAt, now, now) + return err +} + +func (r Repository) DeletePluginFile(ctx context.Context, pluginID, namespace, name string) error { + _, err := r.db.ExecContext(ctx, `DELETE FROM plugin_files WHERE plugin_id = ? AND namespace = ? AND path = ?`, pluginID, namespace, name) + return err +} + +func (r Repository) ListPluginFiles(ctx context.Context, pluginID string) ([]PluginFileSummary, error) { + query := `SELECT plugin_id, namespace, path, data_class, size_bytes, expires_at, updated_at, readonly FROM plugin_files` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY updated_at DESC, namespace ASC, path ASC` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var records []PluginFileSummary + for rows.Next() { + var record PluginFileSummary + var readonly int + if err := rows.Scan(&record.PluginID, &record.Namespace, &record.Path, &record.DataClass, &record.SizeBytes, &record.ExpiresAt, &record.UpdatedAt, &readonly); err != nil { + return nil, err + } + record.Readonly = readonly != 0 + records = append(records, record) + } + return records, rows.Err() +} + +func (r Repository) PluginFileUsage(ctx context.Context, pluginID string) (int64, error) { + row := r.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(size_bytes), 0) FROM plugin_files WHERE plugin_id = ?`, pluginID) + var total int64 + return total, row.Scan(&total) +} + +func (r Repository) SaveDiagnostic(ctx context.Context, pluginID, path string, size int64, sections []string) (DiagnosticPackageSummary, error) { + sectionsJSON, err := marshalDefaultObject(sections) + if err != nil { + return DiagnosticPackageSummary{}, err + } + now := r.now().Unix() + result, err := r.db.ExecContext(ctx, ` +INSERT INTO plugin_diagnostics(plugin_id, path, size_bytes, sections_json, created_at) +VALUES (?, ?, ?, ?, ?)`, pluginID, path, size, sectionsJSON, now) + if err != nil { + return DiagnosticPackageSummary{}, err + } + _, _ = result.LastInsertId() + return DiagnosticPackageSummary{PluginID: pluginID, CreatedAt: now, SizeBytes: size, Sections: sections}, nil +} + +func (r Repository) ListDiagnostics(ctx context.Context, pluginID string, limit int) ([]DiagnosticPackageSummary, error) { + if limit <= 0 { + limit = 20 + } + query := `SELECT plugin_id, size_bytes, sections_json, created_at FROM plugin_diagnostics` + var args []any + if pluginID != "" { + query += ` WHERE plugin_id = ?` + args = append(args, pluginID) + } + query += ` ORDER BY created_at DESC, id DESC LIMIT ?` + args = append(args, limit) + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + var records []DiagnosticPackageSummary + for rows.Next() { + var record DiagnosticPackageSummary + var sectionsJSON string + if err := rows.Scan(&record.PluginID, &record.SizeBytes, §ionsJSON, &record.CreatedAt); err != nil { + return nil, err + } + _ = json.Unmarshal([]byte(defaultJSONArray(sectionsJSON)), &record.Sections) + records = append(records, record) + } + return records, rows.Err() +} + func (r Repository) DispatchPlan(ctx context.Context) (DispatchPlan, error) { plugins, err := r.ListPlugins(ctx) if err != nil { diff --git a/internal/pluginmanager/types.go b/internal/pluginmanager/types.go index c18d770..ea1433f 100644 --- a/internal/pluginmanager/types.go +++ b/internal/pluginmanager/types.go @@ -1,6 +1,7 @@ package pluginmanager import ( + "context" "encoding/json" "errors" "net" @@ -89,7 +90,15 @@ const ( DefaultExtractedMaxBytes = 256 * 1024 * 1024 DefaultNonRuntimeMaxBytes = 16 * 1024 * 1024 DefaultInitialWriteTimeout = time.Second + DefaultExternalTimeout = 5 * time.Second DefaultBuildLogMaxBytes = 64 * 1024 + DefaultEventQueueLimit = 1000 + DefaultEventRecentLimit = 1000 + DefaultLabelValueMaxBytes = 64 + DefaultPluginDataQuota = 16 * 1024 * 1024 + DefaultPluginDataKeyLimit = 256 * 1024 + DefaultPluginFileQuota = 32 * 1024 * 1024 + DefaultLogRecentLimit = 500 ) var ( @@ -117,6 +126,12 @@ type Manifest struct { RuntimeLimits RuntimeLimits `json:"runtime_limits"` ConfigSchema json.RawMessage `json:"config_schema"` Secrets []SecretSpec `json:"secrets,omitempty"` + Events []EventSpec `json:"events,omitempty"` + CustomMetrics []MetricSpec `json:"custom_metrics,omitempty"` + BackgroundTasks []TaskSpec `json:"background_tasks,omitempty"` + ExternalDeps []ExternalSpec `json:"external_dependencies,omitempty"` + DataStores []DataStoreSpec `json:"data_stores,omitempty"` + FileStores []FileStoreSpec `json:"file_stores,omitempty"` SupplyChain json.RawMessage `json:"supply_chain"` } @@ -162,9 +177,66 @@ type SecretRotation struct { Reload string `json:"reload,omitempty"` } +type EventSpec struct { + Name string `json:"name"` + Fields []string `json:"fields,omitempty"` +} + +type MetricSpec struct { + Name string `json:"name"` + Type string `json:"type,omitempty"` + Labels []string `json:"labels,omitempty"` +} + +type TaskSpec struct { + ID string `json:"id"` + Name string `json:"name,omitempty"` + Mode string `json:"mode,omitempty"` + Interval string `json:"interval,omitempty"` + RunOnStart bool `json:"run_on_start,omitempty"` + Jitter string `json:"jitter,omitempty"` + Timeout string `json:"timeout,omitempty"` + Manual bool `json:"manual,omitempty"` + RequireRole string `json:"require_role,omitempty"` +} + +type ExternalSpec struct { + Name string `json:"name"` + Endpoint string `json:"endpoint"` + Purpose string `json:"purpose,omitempty"` + Required bool `json:"required,omitempty"` + Timeout string `json:"timeout,omitempty"` + Retry int `json:"retry,omitempty"` + FailPolicy string `json:"fail_policy,omitempty"` + DataClasses []string `json:"data_classes,omitempty"` + Traceparent bool `json:"traceparent,omitempty"` +} + +type DataStoreSpec struct { + Name string `json:"name"` + SchemaVersion int `json:"schema_version,omitempty"` + DataClass string `json:"data_class,omitempty"` + QuotaBytes int64 `json:"quota_bytes,omitempty"` + Retention string `json:"retention,omitempty"` + Exportable bool `json:"exportable,omitempty"` +} + +type FileStoreSpec struct { + Namespace string `json:"namespace"` + DataClass string `json:"data_class,omitempty"` + QuotaBytes int64 `json:"quota_bytes,omitempty"` + Retention string `json:"retention,omitempty"` + Readonly bool `json:"readonly,omitempty"` +} + type CapabilitySummary struct { UpstreamConnect UpstreamConnectCapability `json:"upstream_connect,omitempty"` Minecraft *MinecraftCapability `json:"minecraft,omitempty"` + Events []EventSpec `json:"events,omitempty"` + CustomMetrics []MetricSpec `json:"custom_metrics,omitempty"` + ExternalDeps []ExternalSpec `json:"external_dependencies,omitempty"` + DataStores []DataStoreSpec `json:"data_stores,omitempty"` + FileStores []FileStoreSpec `json:"file_stores,omitempty"` Raw json.RawMessage `json:"raw,omitempty"` } @@ -604,6 +676,162 @@ type DispatchHandlerSummary struct { ProxyBytesIn uint64 `json:"proxy_bytes_in"` ProxyBytesOut uint64 `json:"proxy_bytes_out"` ProxyDurationMS uint64 `json:"proxy_duration_ms"` + DurationCount uint64 `json:"duration_count"` + DurationSumMS uint64 `json:"duration_sum_ms"` + DurationMaxMS uint64 `json:"duration_max_ms"` +} + +type OperationsSnapshot struct { + PluginID string `json:"plugin_id,omitempty"` + UpdatedAt int64 `json:"updated_at"` + Handlers []DispatchHandlerSummary `json:"handlers"` + Builds []BuildMetricSummary `json:"builds"` + Events []EventSummary `json:"events"` + CustomMetrics []CustomMetricSummary `json:"custom_metrics"` + Logs []LogSummary `json:"logs"` + Traces []TraceSummary `json:"traces"` + BackgroundTasks []BackgroundTaskSummary `json:"background_tasks"` + PluginData []PluginDataSummary `json:"plugin_data"` + PluginFiles []PluginFileSummary `json:"plugin_files"` + ExternalDependencies []ExternalDependencySummary `json:"external_dependencies"` + GC []GCCandidate `json:"gc,omitempty"` + EventQueue EventQueueSummary `json:"event_queue"` + Diagnostics []DiagnosticPackageSummary `json:"diagnostics,omitempty"` +} + +type BuildMetricSummary struct { + PluginID string `json:"plugin_id"` + BuildID int64 `json:"build_id"` + Status string `json:"status"` + DurationMS int64 `json:"duration_ms"` + Failed bool `json:"failed"` + ErrorRedacted string `json:"error_redacted,omitempty"` + CreatedAt int64 `json:"created_at"` +} + +type EventSummary struct { + PluginID string `json:"plugin_id"` + Name string `json:"name"` + Count uint64 `json:"count"` + Dropped uint64 `json:"dropped"` + DeadLetters uint64 `json:"dead_letters"` + Fields map[string]string `json:"fields,omitempty"` + LastSeenAt int64 `json:"last_seen_at"` +} + +type EventQueueSummary struct { + Limit int `json:"limit"` + Queued int `json:"queued"` + Dropped uint64 `json:"dropped"` + DeadLetters uint64 `json:"dead_letters"` +} + +type CustomMetricSummary struct { + PluginID string `json:"plugin_id"` + Name string `json:"name"` + Type string `json:"type"` + Count uint64 `json:"count"` + LastValue float64 `json:"last_value"` + Labels map[string]string `json:"labels,omitempty"` + LastSeenAt int64 `json:"last_seen_at"` +} + +type LogSummary struct { + PluginID string `json:"plugin_id"` + Level string `json:"level"` + Message string `json:"message"` + Fields map[string]string `json:"fields,omitempty"` + TraceID string `json:"trace_id,omitempty"` + ConnectionID string `json:"connection_id,omitempty"` + CreatedAt int64 `json:"created_at"` +} + +type TraceSummary struct { + PluginID string `json:"plugin_id"` + TraceID string `json:"trace_id"` + ConnectionID string `json:"connection_id"` + HandlerID string `json:"handler_id,omitempty"` + Operation string `json:"operation"` + Status string `json:"status"` + DurationMS int64 `json:"duration_ms"` + CreatedAt int64 `json:"created_at"` +} + +type BackgroundTaskSummary struct { + PluginID string `json:"plugin_id"` + TaskID string `json:"task_id"` + Name string `json:"name"` + Mode string `json:"mode"` + IntervalMS int64 `json:"interval_ms"` + RunOnStart bool `json:"run_on_start"` + TimeoutMS int64 `json:"timeout_ms"` + Manual bool `json:"manual"` + Running bool `json:"running"` + LastRunAt int64 `json:"last_run_at"` + NextRunAt int64 `json:"next_run_at"` + LastDurationMS int64 `json:"last_duration_ms"` + LastError string `json:"last_error,omitempty"` + Skipped uint64 `json:"skipped"` + ConsecutiveFailures uint64 `json:"consecutive_failures"` +} + +type PluginDataSummary struct { + PluginID string `json:"plugin_id"` + Key string `json:"key"` + SchemaVersion int `json:"schema_version"` + DataClass string `json:"data_class"` + Exportable bool `json:"exportable"` + SizeBytes int64 `json:"size_bytes"` + ExpiresAt int64 `json:"expires_at"` + UpdatedAt int64 `json:"updated_at"` +} + +type PluginDataRecord struct { + PluginDataSummary + Value []byte `json:"-"` +} + +type PluginFileSummary struct { + PluginID string `json:"plugin_id"` + Namespace string `json:"namespace"` + Path string `json:"path"` + DataClass string `json:"data_class"` + SizeBytes int64 `json:"size_bytes"` + ExpiresAt int64 `json:"expires_at"` + UpdatedAt int64 `json:"updated_at"` + Orphaned bool `json:"orphaned"` + Readonly bool `json:"readonly"` +} + +type PluginFileRecord struct { + PluginFileSummary +} + +type ExternalDependencySummary struct { + PluginID string `json:"plugin_id"` + Name string `json:"name"` + Endpoint string `json:"endpoint"` + Purpose string `json:"purpose"` + Required bool `json:"required"` + FailPolicy string `json:"fail_policy"` + DataClasses []string `json:"data_classes,omitempty"` + Requests uint64 `json:"requests"` + Errors uint64 `json:"errors"` + Inflight int64 `json:"inflight"` + DurationCount uint64 `json:"duration_count"` + DurationSumMS uint64 `json:"duration_sum_ms"` + CircuitState string `json:"circuit_state"` + ConsecutiveFailures uint64 `json:"consecutive_failures"` + RecentError string `json:"recent_error,omitempty"` + LastStatus string `json:"last_status,omitempty"` + LastSeenAt int64 `json:"last_seen_at"` +} + +type DiagnosticPackageSummary struct { + PluginID string `json:"plugin_id"` + CreatedAt int64 `json:"created_at"` + SizeBytes int64 `json:"size_bytes"` + Sections []string `json:"sections"` } type UpstreamResult struct { @@ -621,14 +849,16 @@ type Gateway struct { handleConn func(net.Conn) wg *sync.WaitGroup hooks map[string]any + ops *PluginOperations } -func NewGateway(pluginID string, handleConn func(net.Conn), wg *sync.WaitGroup) *Gateway { +func NewGateway(pluginID string, handleConn func(net.Conn), wg *sync.WaitGroup, ops *PluginOperations) *Gateway { return &Gateway{ PluginID: pluginID, handleConn: handleConn, wg: wg, hooks: make(map[string]any), + ops: ops, } } @@ -658,6 +888,55 @@ func (g *Gateway) ExitWaitGroup() *sync.WaitGroup { return g.wg } +func (g *Gateway) EmitEvent(ctx context.Context, name string, fields map[string]string) error { + if g.ops == nil { + return nil + } + return g.ops.EmitEvent(ctx, name, fields) +} + +func (g *Gateway) ObserveMetric(ctx context.Context, name string, value float64, labels map[string]string) error { + if g.ops == nil { + return nil + } + return g.ops.ObserveMetric(ctx, name, value, labels) +} + +func (g *Gateway) Logger() api.Logger { + if g.ops == nil { + return noopOperationsLogger{} + } + return g.ops.Logger() +} + +func (g *Gateway) DataStore() api.DataStore { + if g.ops == nil { + return noopOperationsDataStore{} + } + return g.ops.DataStore() +} + +func (g *Gateway) FileStore() api.FileStore { + if g.ops == nil { + return noopOperationsFileStore{} + } + return g.ops.FileStore() +} + +func (g *Gateway) ExternalClient(name string) api.ExternalClient { + if g.ops == nil { + return noopOperationsExternalClient{} + } + return g.ops.ExternalClient(name) +} + +func (g *Gateway) RegisterBackgroundTask(task api.BackgroundTask) error { + if g.ops == nil { + return nil + } + return g.ops.RegisterBackgroundTask(task) +} + func (g *Gateway) LegacyUpstreamHandler() (api.HookHandler[func(net.Conn, string) bool, func(net.Conn, string) (net.Conn, error)], bool) { handler, ok := g.hooks[api.HookUpstream.Key()].(api.HookHandler[func(net.Conn, string) bool, func(net.Conn, string) (net.Conn, error)]) return handler, ok diff --git a/plugin/api/api.go b/plugin/api/api.go index f5f5aae..48e16db 100644 --- a/plugin/api/api.go +++ b/plugin/api/api.go @@ -1,8 +1,12 @@ package api import ( + "context" + "io" "net" + "net/http" "sync" + "time" ) type ( @@ -67,6 +71,85 @@ type ( ExitWaitGroup() *sync.WaitGroup Hook(hook string, handler any) error + + EmitEvent(ctx context.Context, name string, fields map[string]string) error + ObserveMetric(ctx context.Context, name string, value float64, labels map[string]string) error + Logger() Logger + DataStore() DataStore + FileStore() FileStore + ExternalClient(name string) ExternalClient + RegisterBackgroundTask(task BackgroundTask) error + } + + EventSchema struct { + Name string `json:"name"` + Fields []string `json:"fields,omitempty"` + } + + CustomMetricSchema struct { + Name string `json:"name"` + Type string `json:"type,omitempty"` + Labels []string `json:"labels,omitempty"` + } + + Logger interface { + Debug(ctx context.Context, message string, fields map[string]string) + Info(ctx context.Context, message string, fields map[string]string) + Warn(ctx context.Context, message string, fields map[string]string) + Error(ctx context.Context, message string, fields map[string]string) + } + + DataRecord struct { + Key string + Value []byte + SchemaVersion int + DataClass string + Exportable bool + Retention time.Duration + } + + DataStore interface { + Put(ctx context.Context, record DataRecord) error + Get(ctx context.Context, key string) (DataRecord, error) + Delete(ctx context.Context, key string) error + } + + FileStore interface { + ResourcePath(name string) (string, error) + Write(ctx context.Context, namespace, name string, data []byte, dataClass string, retention time.Duration) error + Read(ctx context.Context, namespace, name string, maxBytes int64) ([]byte, error) + Delete(ctx context.Context, namespace, name string) error + } + + ExternalRequest struct { + Method string + URL string + Header http.Header + Body io.Reader + Timeout time.Duration + } + + ExternalResponse struct { + StatusCode int + Header http.Header + Body []byte + } + + ExternalClient interface { + DoHTTP(ctx context.Context, req ExternalRequest) (ExternalResponse, error) + DialTCP(ctx context.Context, address string, timeout time.Duration) (net.Conn, error) + HealthCheck(ctx context.Context) error + } + + BackgroundTask struct { + ID string + Name string + Interval time.Duration + RunOnStart bool + Jitter time.Duration + Timeout time.Duration + Manual bool + Run func(context.Context) error } AbstractPlugin struct{} diff --git a/plugin/api/api_test.go b/plugin/api/api_test.go index 3d5b3e9..01290a4 100644 --- a/plugin/api/api_test.go +++ b/plugin/api/api_test.go @@ -1,9 +1,11 @@ package api import ( + "context" "net" "sync" "testing" + "time" ) func TestAbstractPluginDefaults(t *testing.T) { @@ -92,3 +94,67 @@ func (g *recordingGateway) Hook(hook string, handler any) error { g.hooks[hook] = handler return nil } + +func (g *recordingGateway) EmitEvent(context.Context, string, map[string]string) error { + return nil +} + +func (g *recordingGateway) ObserveMetric(context.Context, string, float64, map[string]string) error { + return nil +} + +func (g *recordingGateway) Logger() Logger { + return testLogger{} +} + +func (g *recordingGateway) DataStore() DataStore { + return testDataStore{} +} + +func (g *recordingGateway) FileStore() FileStore { + return testFileStore{} +} + +func (g *recordingGateway) ExternalClient(string) ExternalClient { + return testExternalClient{} +} + +func (g *recordingGateway) RegisterBackgroundTask(BackgroundTask) error { + return nil +} + +type testLogger struct{} + +func (testLogger) Debug(context.Context, string, map[string]string) {} +func (testLogger) Info(context.Context, string, map[string]string) {} +func (testLogger) Warn(context.Context, string, map[string]string) {} +func (testLogger) Error(context.Context, string, map[string]string) {} + +type testDataStore struct{} + +func (testDataStore) Put(context.Context, DataRecord) error { return nil } +func (testDataStore) Get(context.Context, string) (DataRecord, error) { + return DataRecord{}, nil +} +func (testDataStore) Delete(context.Context, string) error { return nil } + +type testFileStore struct{} + +func (testFileStore) ResourcePath(string) (string, error) { return "", nil } +func (testFileStore) Write(context.Context, string, string, []byte, string, time.Duration) error { + return nil +} +func (testFileStore) Read(context.Context, string, string, int64) ([]byte, error) { + return nil, nil +} +func (testFileStore) Delete(context.Context, string, string) error { return nil } + +type testExternalClient struct{} + +func (testExternalClient) DoHTTP(context.Context, ExternalRequest) (ExternalResponse, error) { + return ExternalResponse{}, nil +} +func (testExternalClient) DialTCP(context.Context, string, time.Duration) (net.Conn, error) { + return nil, nil +} +func (testExternalClient) HealthCheck(context.Context) error { return nil } diff --git a/plugin/api/hook.go b/plugin/api/hook.go index 2dc8ddf..bae5f84 100644 --- a/plugin/api/hook.go +++ b/plugin/api/hook.go @@ -14,6 +14,9 @@ var ( ) type ( + ConnectionIDContextKey struct{} + TraceIDContextKey struct{} + HookType[Accept, Handler any] struct { key string }