feat: persist boundary logs (#24812)

Add database persistence to `ReportBoundaryLogs`. On first log for a
session, the handler lazy-creates a `boundary_sessions` row, then
batch-inserts all `BoundaryLog` entries into `boundary_logs`. Structured
logging and usage tracking are preserved. Old boundary clients (no
`session_id`) fall back to log-only mode.

> [!NOTE]
> This PR was authored by Coder Agents.
This commit is contained in:
Sas Swart
2026-06-15 12:34:48 +02:00
committed by GitHub
parent e1c7e61eb9
commit f0ac52e83c
17 changed files with 868 additions and 130 deletions
+2
View File
@@ -237,6 +237,8 @@ func New(opts Options, workspace database.Workspace, agent database.WorkspaceAge
api.BoundaryLogsAPI = &BoundaryLogsAPI{
Log: opts.Log,
Database: opts.Database,
AgentID: opts.AgentID,
WorkspaceID: opts.WorkspaceID,
OwnerID: opts.OwnerID,
TemplateID: workspace.TemplateID,
+162 -1
View File
@@ -2,17 +2,46 @@ package agentapi
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
"github.com/google/uuid"
"golang.org/x/xerrors"
"cdr.dev/slog/v3"
agentproto "github.com/coder/coder/v2/agent/proto"
"github.com/coder/coder/v2/coderd/boundaryusage"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/dbtime"
)
const maxBoundaryLogsPerBatch = 1000
// ErrBatchSizeExceeded matches any BatchSizeExceededError via errors.Is.
var ErrBatchSizeExceeded = xerrors.New("boundary logs batch size exceeded")
// BatchSizeExceededError is returned when a ReportBoundaryLogs request
// exceeds maxBoundaryLogsPerBatch. Match it with errors.As for the sizes,
// or errors.Is(err, ErrBatchSizeExceeded) for the category.
type BatchSizeExceededError struct {
BatchSize int
MaxSize int
}
func (e BatchSizeExceededError) Error() string {
return fmt.Sprintf("batch size %d exceeds maximum of %d", e.BatchSize, e.MaxSize)
}
func (BatchSizeExceededError) Is(target error) bool {
return target == ErrBatchSizeExceeded
}
type BoundaryLogsAPI struct {
Log slog.Logger
Database database.Store
AgentID uuid.UUID
WorkspaceID uuid.UUID
OwnerID uuid.UUID
TemplateID uuid.UUID
@@ -23,8 +52,62 @@ type BoundaryLogsAPI struct {
func (a *BoundaryLogsAPI) ReportBoundaryLogs(ctx context.Context, req *agentproto.ReportBoundaryLogsRequest) (*agentproto.ReportBoundaryLogsResponse, error) {
var allowed, denied int64
if len(req.Logs) == 0 {
a.Log.Debug(ctx, "empty boundary logs request, skipping")
return &agentproto.ReportBoundaryLogsResponse{}, nil
}
if len(req.Logs) > maxBoundaryLogsPerBatch {
return nil, BatchSizeExceededError{BatchSize: len(req.Logs), MaxSize: maxBoundaryLogsPerBatch}
}
now := dbtime.Now()
// Parse session_id if present. Old boundary clients may not send it,
// so a missing or invalid session_id disables DB persistence but
// structured logging and usage tracking still run.
var sessionID uuid.UUID
persistEnabled := false
if raw := req.GetSessionId(); raw != "" {
parsed, parseErr := uuid.Parse(raw)
if parseErr != nil {
a.Log.Warn(ctx, "invalid session_id, persistence disabled for this batch",
slog.F("raw_session_id", raw),
slog.Error(parseErr))
} else {
sessionID = parsed
persistEnabled = true
}
}
if persistEnabled {
// Lazy-create the boundary session on first log arrival.
// If this fails (transient DB error), we continue so that
// logs are still persisted. The session will be created on
// a subsequent batch since every request carries the session
// details.
if sessionErr := a.ensureSession(ctx, sessionID, req.GetConfinedProcessName(), now); sessionErr != nil {
a.Log.Error(ctx, "failed to ensure boundary session",
slog.F("session_id", sessionID.String()),
slog.Error(sessionErr))
}
}
// Collect batch insert params while iterating.
batch := database.InsertBoundaryLogsParams{
SessionID: sessionID,
ID: nil,
SequenceNumber: nil,
CapturedAt: nil,
CreatedAt: nil,
Proto: nil,
Method: nil,
Detail: nil,
MatchedRule: nil,
}
for _, l := range req.Logs {
var logTime time.Time
logTime := now
if l.Time != nil {
logTime = l.Time.AsTime()
}
@@ -45,6 +128,8 @@ func (a *BoundaryLogsAPI) ReportBoundaryLogs(ctx context.Context, req *agentprot
fields := []slog.Field{
slog.F("decision", allowBoolToString(l.Allowed)),
slog.F("session_id", req.SessionId),
slog.F("sequence_number", l.SequenceNumber),
slog.F("workspace_id", a.WorkspaceID.String()),
slog.F("template_id", a.TemplateID.String()),
slog.F("template_version_id", a.TemplateVersionID.String()),
@@ -57,12 +142,35 @@ func (a *BoundaryLogsAPI) ReportBoundaryLogs(ctx context.Context, req *agentprot
}
a.Log.With(fields...).Info(ctx, "boundary_request")
var matchedRule string
if l.Allowed && r.HttpRequest.MatchedRule != "" {
matchedRule = r.HttpRequest.MatchedRule
}
batch.ID = append(batch.ID, uuid.New())
batch.SequenceNumber = append(batch.SequenceNumber, l.SequenceNumber)
batch.CapturedAt = append(batch.CapturedAt, now)
batch.CreatedAt = append(batch.CreatedAt, logTime)
batch.Proto = append(batch.Proto, "http")
batch.Method = append(batch.Method, r.HttpRequest.Method)
batch.Detail = append(batch.Detail, r.HttpRequest.Url)
batch.MatchedRule = append(batch.MatchedRule, matchedRule)
default:
a.Log.Warn(ctx, "unknown resource type",
slog.F("workspace_id", a.WorkspaceID.String()))
}
}
// Batch-insert all collected logs in a single query.
if persistEnabled && len(batch.ID) > 0 {
if insertErr := a.insertLogs(ctx, batch); insertErr != nil {
a.Log.Error(ctx, "failed to insert boundary logs",
slog.F("session_id", sessionID.String()),
slog.F("count", len(batch.ID)),
slog.Error(insertErr))
}
}
if a.BoundaryUsageTracker != nil && (allowed > 0 || denied > 0) {
a.BoundaryUsageTracker.Track(a.WorkspaceID, a.OwnerID, allowed, denied)
}
@@ -70,6 +178,59 @@ func (a *BoundaryLogsAPI) ReportBoundaryLogs(ctx context.Context, req *agentprot
return &agentproto.ReportBoundaryLogsResponse{}, nil
}
// ensureSession creates the boundary_sessions row if it does not
// already exist.
func (a *BoundaryLogsAPI) ensureSession(ctx context.Context, sessionID uuid.UUID, confinedProcess string, now time.Time) error {
if a.Database == nil {
return nil
}
// Check the database in case another replica or reconnection
// already created this session.
_, err := a.Database.GetBoundarySessionByID(ctx, sessionID)
if err == nil {
return nil
}
if !errors.Is(err, sql.ErrNoRows) {
return xerrors.Errorf("check boundary session existence: %w", err)
}
// Session does not exist; create it. started_at is the time
// the first log is received by coderd, per the RFC.
_, err = a.Database.InsertBoundarySession(ctx, database.InsertBoundarySessionParams{
ID: sessionID,
WorkspaceAgentID: a.AgentID,
OwnerID: uuid.NullUUID{UUID: a.OwnerID, Valid: true},
ConfinedProcessName: confinedProcess,
StartedAt: now,
UpdatedAt: now,
})
if err != nil {
// A second coderd replica may receive a batch for this session
// before the first replica has finished inserting it. Both
// attempt the INSERT; the second fails with a primary-key
// unique violation. Treat it as success because the session
// now exists.
if database.IsUniqueViolation(err, database.UniqueBoundarySessionsPkey) {
a.Log.Debug(ctx, "boundary session already created by another replica",
slog.F("session_id", sessionID.String()))
return nil
}
return xerrors.Errorf("insert boundary session: %w", err)
}
return nil
}
// insertLogs persists a batch of boundary log entries.
func (a *BoundaryLogsAPI) insertLogs(ctx context.Context, batch database.InsertBoundaryLogsParams) error {
if a.Database == nil {
return nil
}
_, err := a.Database.InsertBoundaryLogs(ctx, batch)
return err
}
//nolint:revive // This stringifies the boolean argument.
func allowBoolToString(b bool) string {
if b {
+456
View File
@@ -0,0 +1,456 @@
package agentapi_test
import (
"context"
"testing"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/timestamppb"
agentproto "github.com/coder/coder/v2/agent/proto"
"github.com/coder/coder/v2/coderd/agentapi"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/dbgen"
"github.com/coder/coder/v2/coderd/database/dbtestutil"
"github.com/coder/coder/v2/coderd/database/dbtime"
"github.com/coder/coder/v2/testutil"
)
// boundaryFixture holds all database prerequisites for boundary log tests.
type boundaryFixture struct {
DB database.Store
AgentID uuid.UUID
WorkspaceID uuid.UUID
OwnerID uuid.UUID
TemplateID uuid.UUID
TemplateVerID uuid.UUID
}
// newBoundaryFixture creates the full workspace-agent prerequisite chain needed
// by InsertBoundarySession's FK constraint on workspace_agent_id.
func newBoundaryFixture(t *testing.T) *boundaryFixture {
t.Helper()
db, _ := dbtestutil.NewDB(t)
user := dbgen.User(t, db, database.User{})
org := dbgen.Organization(t, db, database.Organization{})
tmpl := dbgen.Template(t, db, database.Template{
OrganizationID: org.ID,
CreatedBy: user.ID,
})
tmplVersion := dbgen.TemplateVersion(t, db, database.TemplateVersion{
TemplateID: uuid.NullUUID{Valid: true, UUID: tmpl.ID},
OrganizationID: org.ID,
CreatedBy: user.ID,
})
workspace := dbgen.Workspace(t, db, database.WorkspaceTable{
OrganizationID: org.ID,
TemplateID: tmpl.ID,
OwnerID: user.ID,
})
job := dbgen.ProvisionerJob(t, db, nil, database.ProvisionerJob{
Type: database.ProvisionerJobTypeWorkspaceBuild,
})
build := dbgen.WorkspaceBuild(t, db, database.WorkspaceBuild{
JobID: job.ID,
WorkspaceID: workspace.ID,
TemplateVersionID: tmplVersion.ID,
})
resource := dbgen.WorkspaceResource(t, db, database.WorkspaceResource{
JobID: build.JobID,
})
agent := dbgen.WorkspaceAgent(t, db, database.WorkspaceAgent{
ResourceID: resource.ID,
})
return &boundaryFixture{
DB: db,
AgentID: agent.ID,
WorkspaceID: workspace.ID,
OwnerID: user.ID,
TemplateID: tmpl.ID,
TemplateVerID: tmplVersion.ID,
}
}
// api returns a new BoundaryLogsAPI backed by this fixture's database.
func (f *boundaryFixture) api(t *testing.T) *agentapi.BoundaryLogsAPI {
return &agentapi.BoundaryLogsAPI{
Log: testutil.Logger(t),
Database: f.DB,
AgentID: f.AgentID,
WorkspaceID: f.WorkspaceID,
OwnerID: f.OwnerID,
TemplateID: f.TemplateID,
TemplateVersionID: f.TemplateVerID,
}
}
// preCreateSession inserts a boundary session directly, bypassing ensureSession,
// to simulate a session created by a prior request or a different coderd replica.
func (f *boundaryFixture) preCreateSession(t *testing.T, sessionID uuid.UUID, process string) {
t.Helper()
_, err := f.DB.InsertBoundarySession(context.Background(), database.InsertBoundarySessionParams{
ID: sessionID,
WorkspaceAgentID: f.AgentID,
ConfinedProcessName: process,
StartedAt: dbtime.Now(),
UpdatedAt: dbtime.Now(),
OwnerID: uuid.NullUUID{UUID: f.OwnerID, Valid: true},
})
require.NoError(t, err, "pre-create boundary session")
}
// addAgent creates another workspace agent in the same workspace chain,
// allowing tests to simulate multiple agents sharing one database.
func (f *boundaryFixture) addAgent(t *testing.T) uuid.UUID {
t.Helper()
job := dbgen.ProvisionerJob(t, f.DB, nil, database.ProvisionerJob{
Type: database.ProvisionerJobTypeWorkspaceBuild,
})
build := dbgen.WorkspaceBuild(t, f.DB, database.WorkspaceBuild{
JobID: job.ID,
WorkspaceID: f.WorkspaceID,
BuildNumber: 2,
TemplateVersionID: f.TemplateVerID,
})
resource := dbgen.WorkspaceResource(t, f.DB, database.WorkspaceResource{
JobID: build.JobID,
})
agent := dbgen.WorkspaceAgent(t, f.DB, database.WorkspaceAgent{
ResourceID: resource.ID,
})
return agent.ID
}
func TestReportBoundaryLogs(t *testing.T) {
t.Parallel()
t.Run("PersistsSessionAndLogs", func(t *testing.T) {
t.Parallel()
// Given: a fresh database and two HTTP log entries (one allowed, one denied).
f := newBoundaryFixture(t)
api := f.api(t)
sessionID := uuid.New()
now := dbtime.Now()
// When: boundary logs are reported.
resp, err := api.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: sessionID.String(),
ConfinedProcessName: "claude-code",
Logs: []*agentproto.BoundaryLog{
{
Allowed: true,
Time: timestamppb.New(now),
SequenceNumber: 0,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "GET",
Url: "https://example.com",
MatchedRule: "domain=example.com",
},
},
},
{
Allowed: false,
Time: timestamppb.New(now),
SequenceNumber: 1,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "POST",
Url: "https://evil.com/exfil",
},
},
},
},
})
// Then: one boundary_sessions row and two boundary_logs rows are written.
require.NoError(t, err)
require.NotNil(t, resp)
sess, err := f.DB.GetBoundarySessionByID(context.Background(), sessionID)
require.NoError(t, err)
require.Equal(t, sessionID, sess.ID)
require.Equal(t, f.AgentID, sess.WorkspaceAgentID)
require.Equal(t, "claude-code", sess.ConfinedProcessName)
logs, err := f.DB.ListBoundaryLogsBySessionID(context.Background(), database.ListBoundaryLogsBySessionIDParams{
SessionID: sessionID,
})
require.NoError(t, err)
require.Len(t, logs, 2)
require.Equal(t, int32(0), logs[0].SequenceNumber)
require.Equal(t, "http", logs[0].Proto)
require.Equal(t, "GET", logs[0].Method)
require.Equal(t, "https://example.com", logs[0].Detail)
require.Equal(t, "domain=example.com", logs[0].MatchedRule.String)
require.Equal(t, int32(1), logs[1].SequenceNumber)
require.Equal(t, "http", logs[1].Proto)
require.Equal(t, "POST", logs[1].Method)
require.Equal(t, "https://evil.com/exfil", logs[1].Detail)
require.Equal(t, "", logs[1].MatchedRule.String)
})
t.Run("SessionAlreadyExistsSameInstance", func(t *testing.T) {
t.Parallel()
// Given: a session created during an earlier batch from the same
// BoundaryLogsAPI instance (e.g. the normal second-and-beyond batch path).
f := newBoundaryFixture(t)
api := f.api(t)
sessionID := uuid.New()
f.preCreateSession(t, sessionID, "claude-code")
// When: a subsequent batch arrives for the same session.
resp, err := api.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: sessionID.String(),
ConfinedProcessName: "claude-code",
Logs: []*agentproto.BoundaryLog{
{
Allowed: true,
Time: timestamppb.New(dbtime.Now()),
SequenceNumber: 5,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "GET",
Url: "https://github.com",
MatchedRule: "domain=github.com",
},
},
},
},
})
// Then: no duplicate session row is created and the new log is persisted.
require.NoError(t, err)
require.NotNil(t, resp)
_, err = f.DB.GetBoundarySessionByID(context.Background(), sessionID)
require.NoError(t, err)
logs, err := f.DB.ListBoundaryLogsBySessionID(context.Background(), database.ListBoundaryLogsBySessionIDParams{
SessionID: sessionID,
})
require.NoError(t, err)
require.Len(t, logs, 1)
require.Equal(t, int32(5), logs[0].SequenceNumber)
})
t.Run("SessionAlreadyExistsDifferentInstance", func(t *testing.T) {
t.Parallel()
// Given: a session created by a first BoundaryLogsAPI instance (first
// coderd replica). A second instance backed by the same database receives
// logs for the same session ID.
f := newBoundaryFixture(t)
api1 := f.api(t)
api2 := f.api(t) // independent struct, simulates a different coderd replica
sessionID := uuid.New()
now := dbtime.Now()
// api1 processes the first batch and creates the session.
_, err := api1.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: sessionID.String(),
ConfinedProcessName: "codex",
Logs: []*agentproto.BoundaryLog{
{
Allowed: true,
Time: timestamppb.New(now),
SequenceNumber: 0,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "GET",
Url: "https://openai.com",
},
},
},
},
})
require.NoError(t, err)
// When: api2 processes a subsequent batch for the same session.
resp, err := api2.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: sessionID.String(),
ConfinedProcessName: "codex",
Logs: []*agentproto.BoundaryLog{
{
Allowed: false,
Time: timestamppb.New(now),
SequenceNumber: 1,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "POST",
Url: "https://pastebin.com",
},
},
},
},
})
// Then: the existing session is reused and both log batches are persisted.
require.NoError(t, err)
require.NotNil(t, resp)
_, err = f.DB.GetBoundarySessionByID(context.Background(), sessionID)
require.NoError(t, err, "session must still exist")
logs, err := f.DB.ListBoundaryLogsBySessionID(context.Background(), database.ListBoundaryLogsBySessionIDParams{
SessionID: sessionID,
})
require.NoError(t, err)
require.Len(t, logs, 2, "logs from both instances must be persisted")
})
t.Run("MissingSessionIDFallsBackToLogOnly", func(t *testing.T) {
t.Parallel()
// Given: a real database and a request with no session_id (old boundary client).
f := newBoundaryFixture(t)
api := f.api(t)
// When: boundary logs are reported without a session_id.
resp, err := api.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
Logs: []*agentproto.BoundaryLog{
{
Allowed: true,
Time: timestamppb.New(dbtime.Now()),
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "GET",
Url: "https://example.com",
},
},
},
},
})
// Then: the request succeeds (log-only mode) and no rows are persisted.
require.NoError(t, err)
require.NotNil(t, resp)
logs, err := f.DB.ListBoundaryLogsBySessionID(context.Background(), database.ListBoundaryLogsBySessionIDParams{
SessionID: uuid.Nil,
})
require.NoError(t, err)
require.Empty(t, logs, "no boundary_logs rows should be persisted without a session_id")
})
t.Run("InvalidSessionIDFallsBackToLogOnly", func(t *testing.T) {
t.Parallel()
// Given: a real database and a request with a session_id that is not a valid UUID.
f := newBoundaryFixture(t)
api := f.api(t)
// When: boundary logs are reported with an invalid session_id.
resp, err := api.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: "not-a-uuid",
Logs: []*agentproto.BoundaryLog{
{
Allowed: true,
Time: timestamppb.New(dbtime.Now()),
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "GET",
Url: "https://example.com",
},
},
},
},
})
// Then: the request succeeds (log-only mode) and no rows are persisted.
require.NoError(t, err)
require.NotNil(t, resp)
logs, err := f.DB.ListBoundaryLogsBySessionID(context.Background(), database.ListBoundaryLogsBySessionIDParams{
SessionID: uuid.Nil,
})
require.NoError(t, err)
require.Empty(t, logs, "no boundary_logs rows should be persisted with an invalid session_id")
})
t.Run("SameSessionIDDifferentAgents", func(t *testing.T) {
t.Parallel()
// Given: two workspace agents in the same workspace, both reporting
// logs with the same session ID. A UUID collision across agents is
// negligible in practice; sessions are namespaced by agent_id at
// query time. The first agent creates the session; the second
// agent's ensureSession hits a unique constraint violation and
// treats it as success.
f := newBoundaryFixture(t)
agent2ID := f.addAgent(t)
api1 := f.api(t)
api2 := &agentapi.BoundaryLogsAPI{
Log: testutil.Logger(t),
Database: f.DB,
AgentID: agent2ID,
WorkspaceID: f.WorkspaceID,
OwnerID: f.OwnerID,
TemplateID: f.TemplateID,
TemplateVersionID: f.TemplateVerID,
}
sessionID := uuid.New()
now := dbtime.Now()
// When: agent1 reports the first batch, creating the session.
_, err := api1.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: sessionID.String(),
ConfinedProcessName: "claude-code",
Logs: []*agentproto.BoundaryLog{
{
Allowed: true,
Time: timestamppb.New(now),
SequenceNumber: 0,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "GET",
Url: "https://example.com",
},
},
},
},
})
require.NoError(t, err)
// When: agent2 reports a batch with the same session ID.
// ensureSession should hit the unique violation and treat it as success.
resp, err := api2.ReportBoundaryLogs(context.Background(), &agentproto.ReportBoundaryLogsRequest{
SessionId: sessionID.String(),
ConfinedProcessName: "claude-code",
Logs: []*agentproto.BoundaryLog{
{
Allowed: false,
Time: timestamppb.New(now),
SequenceNumber: 1,
Resource: &agentproto.BoundaryLog_HttpRequest_{
HttpRequest: &agentproto.BoundaryLog_HttpRequest{
Method: "POST",
Url: "https://evil.com/exfil",
},
},
},
},
})
// Then: both agents' logs are persisted under the same session.
require.NoError(t, err)
require.NotNil(t, resp)
sess, err := f.DB.GetBoundarySessionByID(context.Background(), sessionID)
require.NoError(t, err)
require.Equal(t, f.AgentID, sess.WorkspaceAgentID, "session belongs to the first agent that created it")
logs, err := f.DB.ListBoundaryLogsBySessionID(context.Background(), database.ListBoundaryLogsBySessionIDParams{
SessionID: sessionID,
})
require.NoError(t, err)
require.Len(t, logs, 2, "logs from both agents must be persisted")
})
}