Skip to content

Commit b4e8517

Browse files
juicycleffjuicycleff
committed
Add workload service implementation with tests and template support
- Introduced `streaming_test.go` to test fan-in log and health functionalities, ensuring proper handling of goroutine cancellations and replica management. - Created `template_reader.go` to adapt workload service for template operations, allowing for workload spec reading without direct package imports. - Implemented `transitions.go` and `transitions_test.go` to define and validate workload state transitions, ensuring recoverability from failed states. - Added `types.go` to define data structures for workload creation, updates, and listing, including service specifications and labels. - Developed `workload.go` to encapsulate the workload entity, its lifecycle states, and associated methods for managing workloads.
1 parent e137439 commit b4e8517

245 files changed

Lines changed: 30279 additions & 8872 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎admin/admin.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -107,12 +107,12 @@ type SystemStats struct {
107107
// reports it; nil when the provider doesn't have a meaningful
108108
// geographic identity.
109109
type ProviderStatus struct {
110-
Name string `json:"name"`
111-
Region string `json:"region"`
112-
Location *ProviderLocation `json:"location,omitempty"`
113-
Healthy bool `json:"healthy"`
114-
Instances int `json:"instances"`
115-
Capabilities []string `json:"capabilities"`
110+
Name string `json:"name"`
111+
Region string `json:"region"`
112+
Location *ProviderLocation `json:"location,omitempty"`
113+
Healthy bool `json:"healthy"`
114+
Instances int `json:"instances"`
115+
Capabilities []string `json:"capabilities"`
116116
}
117117

118118
// ProviderLocation mirrors provider.Location at the admin layer so

‎admin/service_impl.go‎

Lines changed: 52 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,15 +16,35 @@ import (
1616
"github.com/xraph/ctrlplane/secrets"
1717
)
1818

19+
// ProviderHealthGetter is the read surface admin needs to surface
20+
// live provider reachability through ListProviders. The
21+
// providerhealth.Cache type satisfies this interface; tests can
22+
// stub it out with anything that returns the same shape.
23+
//
24+
// Defined here (instead of importing providerhealth.Cache directly)
25+
// so admin doesn't depend on a sibling subsystem and so consumers
26+
// can wire any health source they like.
27+
type ProviderHealthGetter interface {
28+
Get(name string) (ProviderHealthSnapshot, bool)
29+
}
30+
31+
// ProviderHealthSnapshot mirrors providerhealth.Status's read fields.
32+
type ProviderHealthSnapshot struct {
33+
Healthy bool
34+
Message string
35+
CheckedAt time.Time
36+
}
37+
1938
// service implements the Service interface.
2039
type service struct {
21-
store Store
22-
instStore instance.Store
23-
netStore network.Store
24-
secStore secrets.Store
25-
providers *provider.Registry
26-
events event.Bus
27-
auth auth.Provider
40+
store Store
41+
instStore instance.Store
42+
netStore network.Store
43+
secStore secrets.Store
44+
providers *provider.Registry
45+
events event.Bus
46+
auth auth.Provider
47+
providerHealth ProviderHealthGetter
2848
}
2949

3050
// NewService creates a new admin service.
@@ -48,6 +68,15 @@ func NewService(
4868
}
4969
}
5070

71+
// SetProviderHealth wires a live provider-health source into the
72+
// admin service. Optional — when unset, ListProviders falls back
73+
// to reporting Healthy=true for every registered provider (the
74+
// pre-existing behavior). Wired by app.CtrlPlane.Start once the
75+
// providerhealth cache is constructed.
76+
func (s *service) SetProviderHealth(g ProviderHealthGetter) {
77+
s.providerHealth = g
78+
}
79+
5180
// CreateTenant creates a new tenant.
5281
func (s *service) CreateTenant(ctx context.Context, req CreateTenantRequest) (*Tenant, error) {
5382
claims, err := auth.RequireClaims(ctx)
@@ -390,11 +419,26 @@ func (s *service) ListProviders(ctx context.Context) ([]ProviderStatus, error) {
390419
}
391420
}
392421

422+
// Default to "healthy" when no live source is wired (matches
423+
// the pre-existing behavior). When the providerhealth cache
424+
// is attached, prefer its live read so the dashboard
425+
// reflects actual provider reachability.
426+
healthy := true
427+
if s.providerHealth != nil {
428+
if snap, ok := s.providerHealth.Get(info.Name); ok {
429+
healthy = snap.Healthy
430+
} else {
431+
// Cache hasn't seen this provider yet (cold start
432+
// window). Be honest rather than optimistic.
433+
healthy = false
434+
}
435+
}
436+
393437
statuses = append(statuses, ProviderStatus{
394438
Name: info.Name,
395439
Region: info.Region,
396440
Location: loc,
397-
Healthy: true,
441+
Healthy: healthy,
398442
Capabilities: capStrings,
399443
})
400444
}

‎api/api.go‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,75 @@ func (a *API) RegisterRoutes(router forge.Router) {
5252
a.registerNetworkRoutes(protectRoutes)
5353
a.registerSecretsRoutes(protectRoutes)
5454
a.registerAdminRoutes(protectRoutes)
55+
a.registerStreamRoutes(protectRoutes)
56+
a.registerUsageRoutes(protectRoutes)
57+
}
58+
59+
// registerUsageRoutes wires the live resource-usage endpoints. The
60+
// /usage path namespace is intentionally distinct from the existing
61+
// telemetry /metrics routes — they're a separate concept (the
62+
// in-memory ring buffer fed by docker stats every 10s) used by the
63+
// workspace dashboard cards.
64+
func (a *API) registerUsageRoutes(router forge.Router) {
65+
g := router.Group("/v1", forge.WithGroupTags("usage"))
66+
67+
_ = g.GET("/instances/:instanceId/usage", a.getInstanceUsage,
68+
forge.WithSummary("Get latest instance usage"),
69+
forge.WithDescription("Returns the most recent usage sample stored for the instance."),
70+
forge.WithOperationID("getInstanceUsage"),
71+
)
72+
_ = g.GET("/instances/:instanceId/usage/range", a.getInstanceUsageRange,
73+
forge.WithSummary("Range-query instance usage"),
74+
forge.WithDescription("Returns a downsampled time-series of usage samples (?range=1h&resolution=auto)."),
75+
forge.WithOperationID("getInstanceUsageRange"),
76+
)
77+
_ = g.EventStream("/instances/:instanceId/usage/stream", a.streamInstanceUsage,
78+
forge.WithSummary("Stream instance usage"),
79+
forge.WithDescription("SSE stream of usage samples as the poller produces them."),
80+
forge.WithOperationID("streamInstanceUsage"),
81+
)
82+
83+
_ = g.GET("/workloads/:workloadId/usage/range", a.getWorkloadUsageRange,
84+
forge.WithSummary("Range-query workload usage"),
85+
forge.WithDescription("Returns the workload-aggregated time-series summed across replicas."),
86+
forge.WithOperationID("getWorkloadUsageRange"),
87+
)
88+
_ = g.EventStream("/workloads/:workloadId/usage/stream", a.streamWorkloadUsage,
89+
forge.WithSummary("Stream workload usage"),
90+
forge.WithDescription("SSE stream of per-replica usage samples tagged with replica metadata."),
91+
forge.WithOperationID("streamWorkloadUsage"),
92+
)
93+
}
94+
95+
// registerStreamRoutes wires the Server-Sent Events endpoints for
96+
// per-instance and per-workload health + logs streaming. Mounted on
97+
// the same auth-protected group as the rest of the API; SSE auth
98+
// supports both the existing Authorization header AND a ?token=
99+
// query param promotion (see auth/sse_middleware.go) so browser
100+
// EventSource clients work without custom headers.
101+
func (a *API) registerStreamRoutes(router forge.Router) {
102+
g := router.Group("/v1", forge.WithGroupTags("streaming"))
103+
104+
_ = g.EventStream("/instances/:instanceId/health/stream", a.streamInstanceHealth,
105+
forge.WithSummary("Stream instance health"),
106+
forge.WithDescription("Server-Sent Events stream of HealthResults for the instance."),
107+
forge.WithOperationID("streamInstanceHealth"),
108+
)
109+
_ = g.EventStream("/instances/:instanceId/logs/stream", a.streamInstanceLogs,
110+
forge.WithSummary("Stream instance logs"),
111+
forge.WithDescription("Server-Sent Events stream of structured stdout/stderr log lines."),
112+
forge.WithOperationID("streamInstanceLogs"),
113+
)
114+
_ = g.EventStream("/workloads/:workloadId/health/stream", a.streamWorkloadHealth,
115+
forge.WithSummary("Stream workload health"),
116+
forge.WithDescription("Fans in HealthResults from every replica."),
117+
forge.WithOperationID("streamWorkloadHealth"),
118+
)
119+
_ = g.EventStream("/workloads/:workloadId/logs/stream", a.streamWorkloadLogs,
120+
forge.WithSummary("Stream workload logs"),
121+
forge.WithDescription("Fans in stdout/stderr from every replica, tagged with replica metadata."),
122+
forge.WithOperationID("streamWorkloadLogs"),
123+
)
55124
}
56125

57126
// registerInstanceRoutes registers all instance management routes.

‎api/deploy_handler.go‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,7 @@ import (
1212
func (a *API) deployInstance(ctx forge.Context, req *DeployAPIRequest) (*deploy.Deployment, error) {
1313
domainReq := deploy.DeployRequest{
1414
InstanceID: req.InstanceID,
15-
Image: req.Image,
16-
Env: req.Env,
15+
Services: req.Services,
1716
Strategy: req.Strategy,
1817
Notes: req.Notes,
1918
CommitSHA: req.CommitSHA,

‎api/middleware.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,10 +35,20 @@ func (a *API) authMiddleware(next http.Handler) http.Handler {
3535
// AuthForgeMiddleware returns a Forge middleware that performs bearer token
3636
// authentication. Use this when registering routes into an external Forge
3737
// router (extension mode).
38+
//
39+
// Promotes a `?token=X` query param to `Authorization: Bearer X` when the
40+
// header is absent so SSE clients (browser EventSource can't set custom
41+
// headers, only cookies) still authenticate. The header path always wins
42+
// when both are present.
3843
func (a *API) AuthForgeMiddleware() forge.Middleware {
3944
return func(next forge.Handler) forge.Handler {
4045
return func(ctx forge.Context) error {
4146
token := ctx.Header("Authorization")
47+
if token == "" {
48+
if t := ctx.Request().URL.Query().Get("token"); t != "" {
49+
token = "Bearer " + t
50+
}
51+
}
4252

4353
claims, err := a.cp.Auth().Authenticate(ctx.Context(), token)
4454
if err != nil {

‎api/requests.go‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -67,15 +67,14 @@ type SuspendInstanceRequest struct {
6767
// ---------------------------------------------------------------------------
6868

6969
// DeployAPIRequest binds path + body for POST /v1/instances/:instanceId/deploy.
70-
// Body fields are replicated because deploy.DeployRequest has a conflicting
71-
// InstanceID json tag.
70+
// Services lists only services being changed in this rollout — services
71+
// not listed inherit their snapshot from the prior Release.
7272
type DeployAPIRequest struct {
73-
InstanceID id.ID `description:"Instance identifier" path:"instanceId"`
74-
Image string `description:"Container image" json:"image"`
75-
Env map[string]string `description:"Environment overrides" json:"env,omitempty"`
76-
Strategy string `description:"Deploy strategy" json:"strategy,omitempty"`
77-
Notes string `description:"Deploy notes" json:"notes,omitempty"`
78-
CommitSHA string `description:"Git commit SHA" json:"commit_sha,omitempty"`
73+
InstanceID id.ID `description:"Instance identifier" path:"instanceId"`
74+
Services []provider.ServiceDeploySpec `description:"Per-service deploy spec" json:"services" validate:"required,min=1"`
75+
Strategy string `description:"Deploy strategy" json:"strategy,omitempty"`
76+
Notes string `description:"Deploy notes" json:"notes,omitempty"`
77+
CommitSHA string `description:"Git commit SHA" json:"commit_sha,omitempty"`
7978
}
8079

8180
// ListDeploymentsRequest binds path + query for GET /v1/instances/:instanceId/deployments.

0 commit comments

Comments
 (0)