mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
Michael Suchacz
119f2b1dd9
Limits concurrent chat generation on capped deployments to 5 root chats and 10 delegated subagent chats. The pools are deployment-wide and independent, so delegated work can continue while root capacity is full. The default caps live in AGPL code. Enterprise contributes only a licensing unlock, so unlicensed deployments stay capped and cannot fail open. Licensed deployments are uncapped while Agent Hours usage stays below an explicit hard limit. Deployments without a hard limit remain uncapped, and reaching the Agent Hours allocation only triggers warnings. Admission happens before a worker takes chat ownership. Capped deployments serialize admission across replicas with a transaction-scoped advisory lock and derive active and queued state from current ownership plus fresh runner heartbeats, rather than persisted queue markers or per-replica state. The acquisition query returns a bounded, pool-interleaved candidate set instead of ranking the whole backlog; a migration replaces the acquisition index with a pool-aware one. Refused chats stay running but unowned, and interrupt requests bypass admission so users can stop queued or over-cap chats. The single-chat API derives `queued_for_capacity` from live pool state; list endpoints do not report it. The UI polls that value every 5 seconds while a chat is running and shows a callout when the chat is waiting for capacity. Updates the administrator documentation and deployment-wide Prometheus gauges for active and queued agents. Replica-level values must be aggregated with `max`, not `sum`. > Mux updated this PR on Mike's behalf.
85 lines
2.7 KiB
Go
85 lines
2.7 KiB
Go
package chatd
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
|
|
"github.com/coder/coder/v2/coderd/database"
|
|
)
|
|
|
|
type capacityMetrics struct {
|
|
active *prometheus.GaugeVec
|
|
queued *prometheus.GaugeVec
|
|
}
|
|
|
|
func newCapacityMetrics(registerer prometheus.Registerer) *capacityMetrics {
|
|
m := &capacityMetrics{
|
|
active: prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Namespace: "coderd",
|
|
Subsystem: "chatd",
|
|
Name: "agents_active",
|
|
Help: "Deployment-wide number of chats holding a concurrent-agent capacity slot. Every replica reports the same database-derived value; aggregate with max, not sum.",
|
|
}, []string{"pool"}),
|
|
queued: prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Namespace: "coderd",
|
|
Subsystem: "chatd",
|
|
Name: "agents_queued_for_capacity",
|
|
Help: "Deployment-wide number of chats waiting for a concurrent-agent capacity slot. Every replica reports the same database-derived value; aggregate with max, not sum.",
|
|
}, []string{"pool"}),
|
|
}
|
|
registerer.MustRegister(m.active, m.queued)
|
|
return m
|
|
}
|
|
|
|
func (w *chatWorker) capacityMetricsLoop(ctx context.Context) {
|
|
ticker := w.opts.Clock.NewTicker(w.opts.CapacityMetricsInterval, "chatworker", "capacity-metrics")
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
w.refreshCapacityMetrics(ctx)
|
|
}
|
|
}
|
|
|
|
func (w *chatWorker) refreshCapacityMetrics(ctx context.Context) {
|
|
active, err := w.opts.Store.CountChatCapacityActiveByPool(ctx, database.CountChatCapacityActiveByPoolParams{
|
|
ExcludeChatID: uuid.Nil,
|
|
StaleSeconds: w.opts.HeartbeatStaleSeconds,
|
|
})
|
|
if err != nil {
|
|
if ctx.Err() == nil {
|
|
w.opts.Logger.Warn(ctx, "chatworker count active capacity chats failed", slogError(err))
|
|
}
|
|
return
|
|
}
|
|
|
|
limits, capped := w.opts.AgentCapacityLimiter.Limits()
|
|
var queuedRoot, queuedSubagent int64
|
|
if capped && (active.ActiveRootCount >= limits.Root || active.ActiveSubagentCount >= limits.Subagent) {
|
|
queued, err := w.opts.Store.CountChatCapacityQueuedByPool(ctx, w.opts.HeartbeatStaleSeconds)
|
|
if err != nil {
|
|
if ctx.Err() == nil {
|
|
w.opts.Logger.Warn(ctx, "chatworker count queued capacity chats failed", slogError(err))
|
|
}
|
|
return
|
|
}
|
|
if active.ActiveRootCount >= limits.Root {
|
|
queuedRoot = queued.QueuedRootCount
|
|
}
|
|
if active.ActiveSubagentCount >= limits.Subagent {
|
|
queuedSubagent = queued.QueuedSubagentCount
|
|
}
|
|
}
|
|
|
|
metrics := w.opts.CapacityMetrics
|
|
metrics.active.WithLabelValues("root").Set(float64(active.ActiveRootCount))
|
|
metrics.active.WithLabelValues("subagent").Set(float64(active.ActiveSubagentCount))
|
|
metrics.queued.WithLabelValues("root").Set(float64(queuedRoot))
|
|
metrics.queued.WithLabelValues("subagent").Set(float64(queuedSubagent))
|
|
}
|