mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-08-31 00:47:15 +02:00
fix: forward resample algorithm to metric store for running jobs
Only archived job data and internal-store node-list data honoured the selected resample algorithm. For running jobs the algorithm was dropped: MetricDataRepository.LoadData had no such parameter, so the memory store fell back to an empty string, which cc-lib's GetResampler maps to LTTB. The external store client was worse - its APIQueryRequest had no ResampleAlgo field at all, and LoadNodeListData accepted the parameter without using it. Add resampleAlgo to the LoadData interface (mirroring LoadNodeListData), forward it from metricdispatch, and set it on both stores' requests. The field is tagged omitempty, so the wire format is unchanged when empty - verify the deployed cc-metric-store accepts it before relying on it there. The REST job endpoints pass a non-zero resolution and therefore do resample, so they now request the configured default instead of an empty string. Add config.ResampleAlgo() for that, since "" is not a neutral value at this layer. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -212,7 +212,7 @@ func TestRestApi(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}}
|
}}
|
||||||
|
|
||||||
metricstore.TestLoadDataCallback = func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int) (schema.JobData, error) {
|
metricstore.TestLoadDataCallback = func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int, resampleAlgo string) (schema.JobData, error) {
|
||||||
return testData, nil
|
return testData, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -513,7 +513,7 @@ func TestStopJobWithReusedJobId(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}}
|
}}
|
||||||
|
|
||||||
metricstore.TestLoadDataCallback = func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int) (schema.JobData, error) {
|
metricstore.TestLoadDataCallback = func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int, resampleAlgo string) (schema.JobData, error) {
|
||||||
return testData, nil
|
return testData, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -309,7 +309,7 @@ func (api *RestAPI) getCompleteJobByID(rw http.ResponseWriter, r *http.Request)
|
|||||||
}
|
}
|
||||||
|
|
||||||
if r.URL.Query().Get("all-metrics") == "true" {
|
if r.URL.Query().Get("all-metrics") == "true" {
|
||||||
data, err = metricdispatch.LoadData(job, nil, scopes, r.Context(), resolution, "")
|
data, err = metricdispatch.LoadData(job, nil, scopes, r.Context(), resolution, config.ResampleAlgo())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cclog.Warnf("REST: error while loading all-metrics job data for JobID %d on %s", job.JobID, job.Cluster)
|
cclog.Warnf("REST: error while loading all-metrics job data for JobID %d on %s", job.JobID, job.Cluster)
|
||||||
return
|
return
|
||||||
@@ -405,7 +405,7 @@ func (api *RestAPI) getJobByID(rw http.ResponseWriter, r *http.Request) {
|
|||||||
resolution = max(resolution, mc.Timestep)
|
resolution = max(resolution, mc.Timestep)
|
||||||
}
|
}
|
||||||
|
|
||||||
data, err := metricdispatch.LoadData(job, metrics, scopes, r.Context(), resolution, "")
|
data, err := metricdispatch.LoadData(job, metrics, scopes, r.Context(), resolution, config.ResampleAlgo())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cclog.Warnf("REST: error while loading job data for JobID %d on %s", job.JobID, job.Cluster)
|
cclog.Warnf("REST: error while loading job data for JobID %d on %s", job.JobID, job.Cluster)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -547,7 +547,7 @@ func TestNatsHandleStopJob(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}}
|
}}
|
||||||
|
|
||||||
metricstore.TestLoadDataCallback = func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int) (schema.JobData, error) {
|
metricstore.TestLoadDataCallback = func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int, resampleAlgo string) (schema.JobData, error) {
|
||||||
return testData, nil
|
return testData, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -193,6 +193,21 @@ func initResampler() {
|
|||||||
// DefaultResamplePolicy is used when no resample policy is configured.
|
// DefaultResamplePolicy is used when no resample policy is configured.
|
||||||
const DefaultResamplePolicy = "medium"
|
const DefaultResamplePolicy = "medium"
|
||||||
|
|
||||||
|
// DefaultResampleAlgo is used when neither the user nor the config selects a
|
||||||
|
// resample algorithm. "average" performs RRDTool-style interval averaging,
|
||||||
|
// which keeps each plotted point a true mean of its interval.
|
||||||
|
const DefaultResampleAlgo = "average"
|
||||||
|
|
||||||
|
// ResampleAlgo returns the configured default resample algorithm, falling back
|
||||||
|
// to DefaultResampleAlgo. Note that an empty string would select LTTB in
|
||||||
|
// cc-lib's resampler, so callers must not pass "" when they mean "the default".
|
||||||
|
func ResampleAlgo() string {
|
||||||
|
if Keys.EnableResampling != nil && Keys.EnableResampling.DefaultAlgo != "" {
|
||||||
|
return Keys.EnableResampling.DefaultAlgo
|
||||||
|
}
|
||||||
|
return DefaultResampleAlgo
|
||||||
|
}
|
||||||
|
|
||||||
// TargetPointsForPolicy returns the target number of data points for a resample
|
// TargetPointsForPolicy returns the target number of data points for a resample
|
||||||
// policy. This is the single source of truth: it feeds both the requested
|
// policy. This is the single source of truth: it feeds both the requested
|
||||||
// resolution (via metricdispatch.ComputeResolution) and the resampler's
|
// resolution (via metricdispatch.ComputeResolution) and the resampler's
|
||||||
|
|||||||
@@ -23,13 +23,14 @@
|
|||||||
// - Running jobs: 2 minutes (data changes frequently)
|
// - Running jobs: 2 minutes (data changes frequently)
|
||||||
// - Completed jobs: 5 hours (data is static)
|
// - Completed jobs: 5 hours (data is static)
|
||||||
//
|
//
|
||||||
// The cache key is based on job ID, state, requested metrics, scopes, and resolution.
|
// The cache key is based on job ID, state, requested metrics, scopes, resolution,
|
||||||
|
// and resample algorithm.
|
||||||
//
|
//
|
||||||
// # Usage
|
// # Usage
|
||||||
//
|
//
|
||||||
// The primary entry point is LoadData, which automatically handles both running and archived jobs:
|
// The primary entry point is LoadData, which automatically handles both running and archived jobs:
|
||||||
//
|
//
|
||||||
// jobData, err := metricdispatch.LoadData(job, metrics, scopes, ctx, resolution)
|
// jobData, err := metricdispatch.LoadData(job, metrics, scopes, ctx, resolution, resampleAlgo)
|
||||||
// if err != nil {
|
// if err != nil {
|
||||||
// // Handle error
|
// // Handle error
|
||||||
// }
|
// }
|
||||||
@@ -115,7 +116,7 @@ func LoadData(job *schema.Job,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
jd, err = ms.LoadData(job, metrics, scopes, ctx, resolution)
|
jd, err = ms.LoadData(job, metrics, scopes, ctx, resolution, resampleAlgo)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if len(jd.Metrics) != 0 {
|
if len(jd.Metrics) != 0 {
|
||||||
cclog.Warnf("partial error loading metrics from store for job %d (user: %s, project: %s, cluster: %s-%s): %s",
|
cclog.Warnf("partial error loading metrics from store for job %d (user: %s, project: %s, cluster: %s-%s): %s",
|
||||||
|
|||||||
@@ -24,7 +24,8 @@ type MetricDataRepository interface {
|
|||||||
metrics []string,
|
metrics []string,
|
||||||
scopes []schema.MetricScope,
|
scopes []schema.MetricScope,
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
resolution int) (schema.JobData, error)
|
resolution int,
|
||||||
|
resampleAlgo string) (schema.JobData, error)
|
||||||
|
|
||||||
// Return a map of metrics to a map of nodes to the metric statistics of the job. node scope only.
|
// Return a map of metrics to a map of nodes to the metric statistics of the job. node scope only.
|
||||||
LoadStats(job *schema.Job,
|
LoadStats(job *schema.Job,
|
||||||
|
|||||||
@@ -81,13 +81,14 @@ type CCMetricStore struct {
|
|||||||
// APIQueryRequest represents a request to the cc-metric-store query API.
|
// APIQueryRequest represents a request to the cc-metric-store query API.
|
||||||
// It supports both explicit queries and "for-all-nodes" bulk queries.
|
// It supports both explicit queries and "for-all-nodes" bulk queries.
|
||||||
type APIQueryRequest struct {
|
type APIQueryRequest struct {
|
||||||
Cluster string `json:"cluster"` // Target cluster name
|
Cluster string `json:"cluster"` // Target cluster name
|
||||||
Queries []APIQuery `json:"queries"` // Explicit list of metric queries
|
Queries []APIQuery `json:"queries"` // Explicit list of metric queries
|
||||||
ForAllNodes []string `json:"for-all-nodes"` // Metrics to query for all nodes
|
ForAllNodes []string `json:"for-all-nodes"` // Metrics to query for all nodes
|
||||||
From int64 `json:"from"` // Start time (Unix timestamp)
|
From int64 `json:"from"` // Start time (Unix timestamp)
|
||||||
To int64 `json:"to"` // End time (Unix timestamp)
|
To int64 `json:"to"` // End time (Unix timestamp)
|
||||||
WithStats bool `json:"with-stats"` // Include min/avg/max statistics
|
WithStats bool `json:"with-stats"` // Include min/avg/max statistics
|
||||||
WithData bool `json:"with-data"` // Include time series data points
|
WithData bool `json:"with-data"` // Include time series data points
|
||||||
|
ResampleAlgo string `json:"resample-algo,omitempty"` // Downsampling algorithm ("lttb", "average", "simple"); empty = server default
|
||||||
}
|
}
|
||||||
|
|
||||||
// APIQuery specifies a single metric query with optional scope filtering.
|
// APIQuery specifies a single metric query with optional scope filtering.
|
||||||
@@ -231,6 +232,7 @@ func (ccms *CCMetricStore) LoadData(
|
|||||||
scopes []schema.MetricScope,
|
scopes []schema.MetricScope,
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
resolution int,
|
resolution int,
|
||||||
|
resampleAlgo string,
|
||||||
) (schema.JobData, error) {
|
) (schema.JobData, error) {
|
||||||
queries, assignedScope, err := ccms.buildQueries(job, metrics, scopes, resolution)
|
queries, assignedScope, err := ccms.buildQueries(job, metrics, scopes, resolution)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -245,12 +247,13 @@ func (ccms *CCMetricStore) LoadData(
|
|||||||
}
|
}
|
||||||
|
|
||||||
req := APIQueryRequest{
|
req := APIQueryRequest{
|
||||||
Cluster: job.Cluster,
|
Cluster: job.Cluster,
|
||||||
From: job.StartTime,
|
From: job.StartTime,
|
||||||
To: job.StartTime + int64(job.Duration),
|
To: job.StartTime + int64(job.Duration),
|
||||||
Queries: queries,
|
Queries: queries,
|
||||||
WithStats: true,
|
WithStats: true,
|
||||||
WithData: true,
|
WithData: true,
|
||||||
|
ResampleAlgo: resampleAlgo,
|
||||||
}
|
}
|
||||||
|
|
||||||
resBody, err := ccms.doRequest(ctx, &req)
|
resBody, err := ccms.doRequest(ctx, &req)
|
||||||
@@ -632,12 +635,13 @@ func (ccms *CCMetricStore) LoadNodeListData(
|
|||||||
}
|
}
|
||||||
|
|
||||||
req := APIQueryRequest{
|
req := APIQueryRequest{
|
||||||
Cluster: cluster,
|
Cluster: cluster,
|
||||||
Queries: queries,
|
Queries: queries,
|
||||||
From: from.Unix(),
|
From: from.Unix(),
|
||||||
To: to.Unix(),
|
To: to.Unix(),
|
||||||
WithStats: true,
|
WithStats: true,
|
||||||
WithData: true,
|
WithData: true,
|
||||||
|
ResampleAlgo: resampleAlgo,
|
||||||
}
|
}
|
||||||
|
|
||||||
resBody, err := ccms.doRequest(ctx, &req)
|
resBody, err := ccms.doRequest(ctx, &req)
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ func (ccms *InternalMetricStore) HealthCheck(cluster string,
|
|||||||
|
|
||||||
// TestLoadDataCallback allows tests to override LoadData behavior for testing purposes.
|
// TestLoadDataCallback allows tests to override LoadData behavior for testing purposes.
|
||||||
// When set to a non-nil function, LoadData will call this function instead of the default implementation.
|
// When set to a non-nil function, LoadData will call this function instead of the default implementation.
|
||||||
var TestLoadDataCallback func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int) (schema.JobData, error)
|
var TestLoadDataCallback func(job *schema.Job, metrics []string, scopes []schema.MetricScope, ctx context.Context, resolution int, resampleAlgo string) (schema.JobData, error)
|
||||||
|
|
||||||
// LoadData loads metric data for a specific job with automatic scope transformation.
|
// LoadData loads metric data for a specific job with automatic scope transformation.
|
||||||
//
|
//
|
||||||
@@ -81,9 +81,10 @@ func (ccms *InternalMetricStore) LoadData(
|
|||||||
scopes []schema.MetricScope,
|
scopes []schema.MetricScope,
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
resolution int,
|
resolution int,
|
||||||
|
resampleAlgo string,
|
||||||
) (schema.JobData, error) {
|
) (schema.JobData, error) {
|
||||||
if TestLoadDataCallback != nil {
|
if TestLoadDataCallback != nil {
|
||||||
return TestLoadDataCallback(job, metrics, scopes, ctx, resolution)
|
return TestLoadDataCallback(job, metrics, scopes, ctx, resolution, resampleAlgo)
|
||||||
}
|
}
|
||||||
|
|
||||||
queries, assignedScope, err := buildQueries(job, metrics, scopes, int64(resolution))
|
queries, assignedScope, err := buildQueries(job, metrics, scopes, int64(resolution))
|
||||||
@@ -99,12 +100,13 @@ func (ccms *InternalMetricStore) LoadData(
|
|||||||
}
|
}
|
||||||
|
|
||||||
req := APIQueryRequest{
|
req := APIQueryRequest{
|
||||||
Cluster: job.Cluster,
|
Cluster: job.Cluster,
|
||||||
From: job.StartTime,
|
From: job.StartTime,
|
||||||
To: job.StartTime + int64(job.Duration),
|
To: job.StartTime + int64(job.Duration),
|
||||||
Queries: queries,
|
Queries: queries,
|
||||||
WithStats: true,
|
WithStats: true,
|
||||||
WithData: true,
|
WithData: true,
|
||||||
|
ResampleAlgo: resampleAlgo,
|
||||||
}
|
}
|
||||||
|
|
||||||
resBody, err := FetchData(req)
|
resBody, err := FetchData(req)
|
||||||
|
|||||||
Reference in New Issue
Block a user