Documentation
¶
Index ¶
- type Coordinator
- func (c *Coordinator) Dispatch(tenantID, requestID string, spec *querierv1.SelectMergeStacktracesRequest)
- func (c *Coordinator) HasCapacity(tenantID string) bool
- func (c *Coordinator) PollQuery(ctx context.Context, tenantID, requestID string) (*Result, error)
- func (c *Coordinator) Submit(ctx context.Context, tenantID string, ...) (string, error)
- type Dispatcher
- type Handler
- type Limits
- type Metadata
- type Result
- type Status
- type Store
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Coordinator ¶
type Coordinator struct {
// contains filtered or unexported fields
}
func NewCoordinator ¶
func NewCoordinator(logger log.Logger, store *Store, limits Limits, next querierv1connect.QuerierServiceHandler, reg prometheus.Registerer) *Coordinator
func (*Coordinator) Dispatch ¶ added in v2.3.0
func (c *Coordinator) Dispatch(tenantID, requestID string, spec *querierv1.SelectMergeStacktracesRequest)
Dispatch implements Store's Dispatcher interface: it (re-)runs a query whose record and spec are already persisted. A declined dispatch (tenant at its concurrency limit) never rolls back the store's claim on the record; it is simply retried by a later adoption scan once the lease expires again.
func (*Coordinator) HasCapacity ¶ added in v2.3.0
func (c *Coordinator) HasCapacity(tenantID string) bool
HasCapacity implements Store's Dispatcher interface: a read-only peek at whether tenantID currently has spare capacity.
func (*Coordinator) PollQuery ¶
PollQuery checks the status of an async query for the given tenant. Returns nil if the request is not found (caller should return NotFound).
func (*Coordinator) Submit ¶ added in v2.3.0
func (c *Coordinator) Submit(ctx context.Context, tenantID string, req *querierv1.SelectMergeStacktracesRequest) (string, error)
Submit reserves the tenant's concurrency slot, strips the Async marker from req, persists the resulting spec as a new in-progress query, and dispatches it in the background. Returns the assigned request ID.
type Dispatcher ¶ added in v2.3.0
type Dispatcher interface {
// HasCapacity is a read-only capacity peek, not a reservation: it makes
// no guarantee that a following Dispatch call will actually run the
// query (Dispatch has its own gate and may still decline).
HasCapacity(tenantID string) bool
Dispatch(tenantID, requestID string, spec *querierv1.SelectMergeStacktracesRequest)
}
Dispatcher re-runs a query that Store has adopted after its owner's lease expired. Implemented by Coordinator; wired into Store via SetDispatcher.
type Handler ¶
type Handler struct {
querierv1connect.QuerierServiceHandler
// contains filtered or unexported fields
}
Handler decorates a QuerierServiceHandler with async query support for SelectMergeStacktraces. All other RPCs pass through to the embedded handler unchanged.
func NewHandler ¶
func NewHandler(logger log.Logger, next querierv1connect.QuerierServiceHandler, coordinator *Coordinator) *Handler
func (*Handler) SelectMergeStacktraces ¶
func (h *Handler) SelectMergeStacktraces( ctx context.Context, req *connect.Request[querierv1.SelectMergeStacktracesRequest], ) (*connect.Response[querierv1.SelectMergeStacktracesResponse], error)
SelectMergeStacktraces honors the request's optional Async field:
- Async == nil or Type == DISABLED: run synchronously, return the wrapped handler's response unchanged (Async stays nil).
- Async.RequestId != "": treat as a poll. All other request fields are ignored. The response carries only the Async metadata, plus the result payload on SUCCESS.
- Async.Type == FORCE with empty RequestId: dispatch in the background and return a response carrying only the Async metadata in IN_PROGRESS.
type Metadata ¶
type Metadata struct {
RequestID string `json:"request_id"`
TenantID string `json:"tenant_id"`
Status Status `json:"status"`
CreatedAt time.Time `json:"created_at"`
LastHeartbeat time.Time `json:"last_heartbeat,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
Owner string `json:"owner,omitempty"`
AdoptionCount int `json:"adoption_count,omitempty"`
}
Metadata describes the state of a single async query. It is persisted as metadata.json alongside the query's result in object storage.
type Result ¶
type Result struct {
Metadata Metadata
Response *querierv1.SelectMergeStacktracesResponse
}
Result bundles a query's Metadata with its decoded Response. Response is only populated when Metadata.Status is StatusSuccess.
type Status ¶
type Status string
Status represents the lifecycle state of an async query.
const ( // StatusInProgress indicates the query is still executing. StatusInProgress Status = "in_progress" // StatusSuccess indicates the query completed and a result is available. StatusSuccess Status = "success" // StatusFailure indicates the query failed; ErrorMessage holds the reason. StatusFailure Status = "failure" )
type Store ¶
Store persists async query state and results in object storage. It also runs as a dskit service that periodically removes expired entries and adopts queries whose owner's lease has expired.
func NewStore ¶
func NewStore(logger log.Logger, bucket objstore.Bucket, reg prometheus.Registerer) *Store
NewStore returns a Store backed by the given bucket. The returned Store is a dskit service that, once started, periodically deletes entries older than the configured TTL and scans for adoptable queries; callers are responsible for starting and stopping it via its embedded Service.
func (*Store) SetDispatcher ¶ added in v2.3.0
func (s *Store) SetDispatcher(d Dispatcher)
SetDispatcher wires the adoption scan to a Dispatcher. Must be called before the Store's Service is started: s.dispatcher is a plain field, never synchronized against running()'s reads of it. A late call is a programming error, so it panics rather than risk an unsynchronized write racing the running service. nil is safe (adoption is then a no-op), e.g. when Store runs standalone or in tests.
Source Files