diff --git a/.github/workflows/step_tests-lint.yml b/.github/workflows/step_tests-lint.yml index 7135ebf7..0487bfa5 100644 --- a/.github/workflows/step_tests-lint.yml +++ b/.github/workflows/step_tests-lint.yml @@ -28,5 +28,5 @@ jobs: - name: Lint uses: golangci/golangci-lint-action@v9 with: - version: v2.11.4 + version: v2.13.2 args: --timeout=5m diff --git a/.golangci.yml b/.golangci.yml index e5b66e96..cfeb0fd2 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -8,11 +8,13 @@ linters: - dupword - err113 - exhaustruct + - exhaustruct_v5 - funlen - gochecknoglobals - gochecknoinits - gocognit - gocritic + - goconst - gocyclo - gosmopolitan - ireturn @@ -57,6 +59,9 @@ linters: ignore-string-values: - cpu - info + ignore-tests: true + ignore-map-keys: true + min: 5 gomoddirectives: replace-allow-list: - github.com/grafana/pyroscope/ebpf @@ -81,8 +86,9 @@ linters: - gosec - noctx - unqueryvet + - goconst # text: "(G705: XSS via taint analysis|G704: SSRF via taint analysis|G703: Path traversal via taint analysis|G702: Command injection via taint analysis|G306: Expect WriteFile permissions to be 0600 or less|G120: Parsing form data without limiting request body size can allow memory exhaustion (use http.MaxBytesReader)|net/http/httptest.NewRequest must not be called. use net/http/httptest.NewRequestWithContext)" - path: scripts/mock_servers + path: scripts - linters: - exhaustive - intrange @@ -98,6 +104,7 @@ formatters: - gofmt - gofumpt - goimports + - swaggo exclusions: generated: lax paths: diff --git a/Makefile b/Makefile index 990ccdf8..cfabb995 100644 --- a/Makefile +++ b/Makefile @@ -54,7 +54,7 @@ PROMU_TEST_CONF ?= .promu/.promu-go-test.yml ifeq ($(CGO_APPS), 1) PROMU_CONF ?= .promu/.promu-cgo.yml pkgs := ./pkg/sqlite3 ./pkg/api/cli \ - ./pkg/api/db ./pkg/api/db/migrator ./pkg/api/helper \ + ./pkg/api/db ./pkg/api/db/migrator \ ./pkg/api/resource ./pkg/api/resource/slurm ./pkg/api/resource/lsf \ ./pkg/api/resource/openstack ./pkg/api/resource/k8s \ ./pkg/api/updater ./pkg/api/updater/tsdb \ @@ -193,12 +193,14 @@ test-e2e: $(PROMTOOL) build pkg/collector/testdata/sys/.unpacked pkg/collector/t ./scripts/e2e-test.sh -s tool-relabel-configs ./scripts/e2e-test.sh -s tool-web-config ./scripts/e2e-test.sh -s cacct-default-format + ./scripts/e2e-test.sh -s cacct-help-format ./scripts/e2e-test.sh -s cacct-long-format ./scripts/e2e-test.sh -s cacct-custom-format ./scripts/e2e-test.sh -s cacct-admin-user ./scripts/e2e-test.sh -s cacct-admin-user-all-users ./scripts/e2e-test.sh -s cacct-forbid-query ./scripts/e2e-test.sh -s cacct-invalid-config + ./scripts/e2e-test.sh -s cacct-bad-config ./scripts/e2e-test.sh -s cacct-tsdata ./scripts/e2e-test.sh -s cacct-tsdata-fail else @@ -290,12 +292,14 @@ test-e2e-update: build pkg/collector/testdata/sys/.unpacked pkg/collector/testda ./scripts/e2e-test.sh -s tool-relabel-configs -u || true ./scripts/e2e-test.sh -s tool-web-config -u || true ./scripts/e2e-test.sh -s cacct-default-format -u || true + ./scripts/e2e-test.sh -s cacct-help-format -u || true ./scripts/e2e-test.sh -s cacct-long-format -u || true ./scripts/e2e-test.sh -s cacct-custom-format -u || true ./scripts/e2e-test.sh -s cacct-admin-user -u || true ./scripts/e2e-test.sh -s cacct-admin-user-all-users -u || true ./scripts/e2e-test.sh -s cacct-forbid-query -u || true ./scripts/e2e-test.sh -s cacct-invalid-config -u || true + ./scripts/e2e-test.sh -s cacct-bad-config -u || true ./scripts/e2e-test.sh -s cacct-tsdata -u || true ./scripts/e2e-test.sh -s cacct-tsdata-fail -u || true else diff --git a/build/config/cacct/cacct.yml b/build/config/cacct/cacct.yml index d4ef4a31..e1a92c1e 100644 --- a/build/config/cacct/cacct.yml +++ b/build/config/cacct/cacct.yml @@ -294,41 +294,96 @@ # # # http_headers: {} -# # To dump the time series data for each metric, this section must be configured. -# # The key name is the name of the metric and value is the PromQL query to get -# # time series data. The placeholder `%s` will be replaced by list of job IDs delimited -# # by `|` which is the syntax expected by TSDB server. +# # List of TSDB queries +# # +# # The keys `name`, `query` and `kind` are mandatory and more over `name` must be +# # unique. The key `title` and `help` can be used to provide human friendly name and +# # help text. For the moment, only `range` kind is supported. # # # # If the TSDB server has been configured with the recording rules generated by `ceems_tool` # # the following queries should work out-of-the-box. # # +# # +# # Available template variables: +# # +# # - UUIDs -> UUIDs string delimited by "|", e.g., 123|345|567 +# # - ScrapeInterval -> Scrape interval of TSDB in time.Duration format, e.g., 15s, 1m +# # - ScrapeIntervalMilli -> Scrape interval of TSDB in milliseconds, e.g., 15000, 60000 +# # - EvaluationInterval -> Evaluation interval of TSDB in time.Duration format, e.g., 15s, 1m +# # - EvaluationIntervalMilli -> Evaluation interval of TSDB in milliseconds, e.g., 15s, 1m +# # - RateInterval -> Rate interval in time.Duration format. It is estimated based on Scrape interval as 4*scrape_interval +# # - Range -> Duration of interval where aggregation is being made in time.Duration format +# # +# # IMPORTANT: Always use backticks around {{.UUIDs}} template variable to escape characters +# # like "[" and "]" which can be found in identifiers of resource managers like LSF. +# # # queries: -# # CPU utilisation -# cpu_usage: uuid:ceems_cpu_usage:ratio_irate{uuid=~"%s"} - -# # CPU Memory utilisation -# cpu_mem_usage: uuid:ceems_cpu_memory_usage:ratio{uuid=~"%s"} - -# # Host power usage in Watts -# host_power_usage: uuid:ceems_host_power_watts:pue{uuid=~"%s"} - -# # Host emissions in g/s -# host_emissions: uuid:ceems_host_emissions_g_s:pue{uuid=~"%s"} - -# # GPU utilization -# avg_gpu_usage: uuid:ceems_gpu_usage:ratio{uuid=~"%s"} - -# # GPU memory utilization -# avg_gpu_mem_usage: uuid:ceems_gpu_memory_usage:ratio{uuid=~"%s"} - -# # GPU power usage in Watts -# gpu_power_usage: uuid:ceems_gpu_power_watts:pue{uuid=~"%s"} - -# # GPU emissions in g/s -# gpu_emissions: uuid:ceems_gpu_emissions_g_s:pue{uuid=~"%s"} - -# # Read IO bytes -# io_read_bytes: irate(ceems_ebpf_read_bytes_total{uuid=~"%s"}[1m]) - -# # Write IO bytes -# io_write_bytes: irate(ceems_ebpf_write_bytes_total{uuid=~"%s"}[1m]) +# # # CPU utilization +# # - name: cpu_usage +# # title: "CPU Usage (%)" +# # help: "Usage of CPUs in %" +# # query: uuid:ceems_cpu_usage:ratio_irate{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # CPU Memory utilization +# # - name: cpu_mem_usage +# # title: "CPU Memory Usage (%)" +# # help: "Ratio of memory used to memory reserved of CPU in %" +# # query: uuid:ceems_cpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # Host power usage in Watts +# # - name: host_power_usage +# # title: "Host Power usage (W)" +# # help: "Instanteous power usage of host in Watts" +# # query: uuid:ceems_host_power_watts:pue{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # Host emissions in g/s +# # - name: host_emissions +# # title: "Host Eq. Emissions Rate (g/s)" +# # help: "Instanteous emissions rate of host in g/s" +# # query: uuid:ceems_host_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # GPU utilization +# # - name: avg_gpu_usage +# # title: "GPU Usage (%)" +# # help: "Usage of GPU in %" +# # query: uuid:ceems_gpu_usage:ratio{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # GPU memory utilization +# # - name: avg_gpu_mem_usage +# # title: "GPU Memory Usage (%)" +# # help: "Ratio of memory used to memory reserved of GPU in %" +# # query: uuid:ceems_gpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # GPU power usage in Watts +# # - name: gpu_power_usage +# # title: "GPU Power Usage (W)" +# # help: "Instanteous Power Usage of GPU in Watts" +# # query: uuid:ceems_gpu_power_watts:pue{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # GPU emissions in g/s +# # - name: gpu_emissions +# # title: "GPU Eq. Emission Rate (g/s)" +# # help: "Instanteous emissions rate of GPU in g/s" +# # query: uuid:ceems_gpu_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} +# # kind: range +# # +# # # Read IO bytes/s +# # - name: io_read_bytes +# # title: "IO Read Bandwidth (b/s)" +# # help: "Instanteous IO read bandwidth in bytes/s" +# # query: irate(ceems_ebpf_read_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) +# # kind: range +# # +# # # Write IO bytes/s +# # - name: io_write_bytes +# # title: "IO Write Bandwidth (b/s)" +# # help: "Instanteous IO write bandwidth in bytes/s" +# # query: irate(ceems_ebpf_write_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) +# # kind: range diff --git a/cmd/cacct/api.go b/cmd/cacct/api.go index cc0ce259..f3597166 100644 --- a/cmd/cacct/api.go +++ b/cmd/cacct/api.go @@ -10,7 +10,6 @@ import ( "log/slog" "net/http" "net/url" - "os" "slices" "strconv" "strings" @@ -35,8 +34,7 @@ func stats( jobs []string, userNames []string, fields []string, - tsData bool, - tsDataOut string, + includeSummaryStats bool, ) ([]models.Unit, []models.Usage, error) { // Add user header to HTTP config userHeaders := http_config.Header{ @@ -93,7 +91,7 @@ func stats( // Even if normal user make a request by requesting // with --user flag, if that user is not in admin list, empty // result will be returned - var unitsReqURL, usageReqURL string + var unitsReqURL, usageReqURL *url.URL if len(userNames) > 0 { // If --user flag does not contain special value "all", add them to the query. @@ -105,14 +103,14 @@ func stats( } } - unitsReqURL = apiURL.JoinPath("/api/v1/units/admin").String() - usageReqURL = apiURL.JoinPath("/api/v1/usage/current/admin").String() + unitsReqURL = apiURL.JoinPath("/api/v1/units/admin") + usageReqURL = apiURL.JoinPath("/api/v1/usage/current/admin") } else { - unitsReqURL = apiURL.JoinPath("/api/v1/units").String() - usageReqURL = apiURL.JoinPath("/api/v1/usage/current").String() + unitsReqURL = apiURL.JoinPath("/api/v1/units") + usageReqURL = apiURL.JoinPath("/api/v1/usage/current") } - logger.Debug("Request to fetch units", "url", unitsReqURL, "units_query", urlValues.Encode()) + logger.Debug("Request to fetch units", "url", unitsReqURL.Redacted(), "units_query", urlValues.Encode()) // If CEEMS URL is available make a API request ctx, cancel := context.WithTimeout(context.Background(), time.Minute) @@ -126,41 +124,27 @@ func stats( return nil, nil, fmt.Errorf("failed to fetch jobs: %w", err) } - // Get all units in the given period - // Always add user field as we will need to get total usage - urlValues.Add("field", "username") - urlValues.Add("field", "num_units") - - // If elapsed is requested we need to get total_time_seconds from usage API resource - if slices.Contains(urlValues["field"], "elapsed") { - urlValues.Add("field", "total_time_seconds") - } + // If summary stats are requested, include them + var usage []models.Usage - logger.Debug("Request to fetch usage", "url", usageReqURL, "usage_query", urlValues.Encode()) - - usage, err := doRequest[models.Usage](ctx, usageReqURL, urlValues, apiClient) - if err != nil { - logger.Error("Failed to fetch usage data from CEEMS API server", "err", err) + if includeSummaryStats { + // Get all units in the given period + // Always add user field as we will need to get total usage + urlValues.Add("field", "username") + urlValues.Add("field", "num_units") - return nil, nil, fmt.Errorf("failed to fetch usage: %w", err) - } - - // If tsData is enabled, get time series data - if tsData { - // If metrics are not configured, return logging a message - if len(config.TSDB.Queries) == 0 { - logger.Warn("TSDB queries not configured") - fmt.Fprintln(os.Stderr, "time series data not available") - - return units, usage, nil + // If elapsed is requested we need to get total_time_seconds from usage API resource + if slices.Contains(urlValues["field"], "elapsed") { + urlValues.Add("field", "total_time_seconds") } - logger.Debug("Fetching time series data from TSDB") + logger.Debug("Request to fetch usage", "url", usageReqURL.Redacted(), "usage_query", urlValues.Encode()) - err := tsdbData(ctx, logger, config, units, tsDataOut) + usage, err = doRequest[models.Usage](ctx, usageReqURL, urlValues, apiClient) if err != nil { - logger.Error("failed to fetch time series data", "err", err) - fmt.Fprintln(os.Stderr, "failed to fetch time series data") + logger.Error("Failed to fetch usage data from CEEMS API server", "err", err) + + return nil, nil, fmt.Errorf("failed to fetch usage: %w", err) } } @@ -168,9 +152,9 @@ func stats( } // doRequest does an API request to CEEMS API server and returns response. -func doRequest[T any](ctx context.Context, reqURL string, urlValues url.Values, client *http.Client) ([]T, error) { +func doRequest[T any](ctx context.Context, reqURL *url.URL, urlValues url.Values, client *http.Client) ([]T, error) { // Make a new request - req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL.String(), nil) if err != nil { return nil, err } diff --git a/cmd/cacct/main.go b/cmd/cacct/main.go index 7c174db8..add25413 100644 --- a/cmd/cacct/main.go +++ b/cmd/cacct/main.go @@ -2,6 +2,7 @@ package main import ( "cmp" + "encoding/json" "errors" "fmt" "io" @@ -17,9 +18,11 @@ import ( "github.com/alecthomas/kingpin/v2" "github.com/ceems-dev/ceems/internal/common" "github.com/ceems-dev/ceems/pkg/api/models" + "github.com/iancoleman/strcase" "github.com/jedib0t/go-pretty/v6/table" "github.com/jedib0t/go-pretty/v6/text" http_config "github.com/prometheus/common/config" + "github.com/prometheus/common/model" "github.com/prometheus/common/promslog" "github.com/prometheus/common/version" "gopkg.in/yaml.v3" @@ -50,7 +53,7 @@ var ( }, "name": { tag: "name", - name: "Name", + name: "name", help: "Name of the job", title: "Name", minW: 5, @@ -58,7 +61,7 @@ var ( }, "account": { tag: "project", - name: "Account", + name: "account", help: "Account name", title: "Account", minW: 5, @@ -112,6 +115,22 @@ var ( minW: 5, maxW: 12, }, + "totaltime": { + tag: "total_time_seconds", + name: "totalTime", + help: "Wall, CPU and GPU times consumed in seconds over the duration of the job", + title: "Total Time(s)", + minW: 5, + maxW: 12, + }, + "allocation": { + tag: "allocation", + name: "allocation", + help: "Resource allocation of CPU, GPU, memory, etc", + title: "Resource Allocation", + minW: 5, + maxW: 12, + }, "state": { tag: "state", name: "state", @@ -196,6 +215,8 @@ var ( "startedat", "endedat", "elapsed", + "totaltime", + "allocation", "state", "cpuusage", "cpumemoryusage", @@ -229,6 +250,7 @@ var ( errConfig = errors.New("unable to get cacct config") errLogFile = errors.New("unable to get open log file") errUser = errors.New("unable to change user context") + errNoUnits = errors.New("no jobs found in the selected period") errInternal = errors.New("internal server error") ) @@ -287,6 +309,52 @@ func (f field) subtitles() []any { // } // ) +const ( + instantQuery = "instant" + rangeQuery = "range" +) + +type TSDBQuery struct { + Name string `yaml:"name"` + Help string `yaml:"help"` + Title string `yaml:"title"` + Query string `yaml:"query"` + Kind string `yaml:"kind"` +} + +// UnmarshalYAML implements the yaml.Unmarshaler interface. +func (q *TSDBQuery) UnmarshalYAML(unmarshal func(any) error) error { + // Set a default config + *q = TSDBQuery{} + + type plain TSDBQuery + + err := unmarshal((*plain)(q)) + if err != nil { + return err + } + + // Check if query is of range or instant + if q.Kind != rangeQuery && q.Kind != instantQuery { + return fmt.Errorf("invalid value %s found for kind. Must be one of %s or %s", q.Kind, instantQuery, rangeQuery) + } + + // Validate config + if q.Name == "" || q.Query == "" || q.Kind == "" { + return errors.New("name, query and kind cannot be empty in entry of queries") + } + + // If title is empty, use same as name + if q.Title == "" { + q.Title = q.Name + } + + // Convert name to camelCase + q.Name = strcase.ToLowerCamel(q.Name) + + return nil +} + // Config contains the cacct configuration settings. type Config struct { API struct { @@ -295,8 +363,16 @@ type Config struct { UserHeaderName string `yaml:"user_header_name"` } `yaml:"ceems_api_server"` TSDB struct { - Web WebConfig `yaml:"web"` - Queries map[string]string `yaml:"queries"` + Web WebConfig `yaml:"web"` + ScrapeInterval model.Duration `yaml:"scrape_interval"` + EvaluationInterval model.Duration `yaml:"evaluation_interval"` + QueryMaxSeries int64 `yaml:"query_max_series"` + QueryMinSamples float64 `yaml:"query_min_samples"` + MaxUnitsForRangeQueries int `yaml:"max_units_for_range_queries"` + QueryTimeout model.Duration `yaml:"query_timeout"` + Queries []TSDBQuery `yaml:"queries"` + rangeQueries []TSDBQuery + instantQueries []TSDBQuery } `yaml:"tsdb"` Logging struct { Enabled bool `yaml:"enabled"` @@ -313,6 +389,10 @@ func (c *Config) UnmarshalYAML(unmarshal func(any) error) error { *c = Config{} c.API.UserHeaderName = "X-Grafana-User" c.Logging.Level = promslog.NewLevel() + c.TSDB.MaxUnitsForRangeQueries = 10 + c.TSDB.QueryMaxSeries = 20 + c.TSDB.QueryMinSamples = 0.5 + c.TSDB.QueryTimeout = model.Duration(time.Minute) err := c.Logging.Level.Set("info") if err != nil { @@ -342,11 +422,37 @@ func (c *Config) UnmarshalYAML(unmarshal func(any) error) error { return err } + // Add instant queries to fieldMap and allFields + for _, q := range c.TSDB.Queries { + nameKey := strings.ToLower(q.Name) + switch q.Kind { + case instantQuery: + fieldMap[nameKey] = &field{ + name: q.Name, + help: q.Help, + title: q.Title, + minW: 2, + maxW: 10, + } + allFields = append(allFields, nameKey) + } + } + return nil } // Validate validates the config. func (c *Config) Validate() error { + // Check there are no duplicate names in range and instant queries + var allQueryNames []string + for _, q := range c.TSDB.Queries { + if slices.Contains(allQueryNames, q.Name) { + return fmt.Errorf("name key %s duplicates found in tsdb.queries %s", q.Name, strings.Join(allQueryNames, ",")) + } + + allQueryNames = append(allQueryNames, q.Name) + } + // If logging is not enabled, nothing to do here if !c.Logging.Enabled { return nil @@ -377,6 +483,19 @@ func (c *Config) Validate() error { return nil } +// SetupTSDBQueries sets up TSDB queries based on CLI args. +func (c *Config) SetupTSDBQueries(instantQueries []string, rangeQueries []string) { + // Add instant queries to fieldMap and allFields + for _, q := range c.TSDB.Queries { + switch { + case q.Kind == instantQuery && slices.Contains(instantQueries, q.Name): + c.TSDB.instantQueries = append(c.TSDB.instantQueries, q) + case q.Kind == rangeQuery && slices.Contains(rangeQueries, q.Name): + c.TSDB.rangeQueries = append(c.TSDB.rangeQueries, q) + } + } +} + // WebConfig contains HTTP related config. type WebConfig struct { URL string `yaml:"url"` @@ -417,9 +536,10 @@ type Response[T any] struct { func main() { var ( - tsData, helpFormat, longFormat bool + helpFormat, longFormat bool htmlOut, csvOut, mdOut bool - tsDataOut string + summaryStats bool + tsDataOut, tsMetrics string accountsFlag, jobsFlag, usersFlag string formatFlag string startTime, endTime string @@ -454,8 +574,11 @@ func main() { "long", fmt.Sprintf("Equivalent to specifying --format=\"%s\".", strings.Join(allFields, ",")), ).Default("false").BoolVar(&longFormat) cacctApp.Flag( - "ts", "Time series data of jobs are saved in CSV format (default: false).", - ).BoolVar(&tsData) + "summary", "Include summary statistics at the end in the results.", + ).Default("true").BoolVar(&summaryStats) + cacctApp.Flag( + "ts.metrics", "Comma separated list of time series metrics. Check available metrics using --helpformat flag.", + ).StringVar(&tsMetrics) cacctApp.Flag( "ts.out-dir", "Directory to save time series data.", ).Default("out").StringVar(&tsDataOut) @@ -474,6 +597,15 @@ func main() { kingpin.Fatalf("failed to parse CLI flags: %v", err) } + // First read the config file to get the queries supplied for TSDB so that we + // can add them to helpformat flag. + // Either setuid or setgid bits must be applied on the app so that + // the config file can be read as the owner of this app + config, err := readConfig(mockConfigPath) + if err != nil { + os.Exit(checkErr(fmt.Errorf("%w: %w", errConfig, err))) + } + // If helpformat, print available fields and return if helpFormat { // First collect keys and sort them @@ -487,6 +619,18 @@ func main() { t.AppendRow(table.Row{fieldMap[k].name, fieldMap[k].help}) } + // Append available time series metrics + t.AppendSeparator() + + t.AppendRow(table.Row{"Available Time Series"}) + t.AppendSeparator() + + for _, q := range config.TSDB.Queries { + if q.Kind == rangeQuery { + t.AppendRow(table.Row{q.Name, q.Help}) + } + } + t.Render() os.Exit(0) @@ -508,11 +652,43 @@ func main() { formatFields = allFields } - var fields []string + var ( + activeInstantQueries []string + activeRangeQueries []string + ) + + // Get active range query names based on CLI args + for _, t := range splitString(tsMetrics, ",") { + for _, q := range config.TSDB.Queries { + if strings.EqualFold(q.Name, t) { + activeRangeQueries = append(activeRangeQueries, q.Name) + } + } + } + + // Get active instant query names based on CLI args + // ALWAYS include uuid in fields + fields := []string{"uuid"} + for _, f := range formatFields { - fields = append(fields, fieldMap[strings.ToLower(f)].tag) + nameKey := strings.ToLower(f) + if field, ok := fieldMap[nameKey]; ok && nameKey != "jobid" { + tag := field.tag + if tag != "" { + fields = append(fields, tag) + } + + for _, q := range config.TSDB.Queries { + if q.Name == field.name && q.Kind == instantQuery { + activeInstantQueries = append(activeInstantQueries, q.Name) + } + } + } } + // Setup queries on config struct + config.SetupTSDBQueries(activeInstantQueries, activeRangeQueries) + // Always add started and ended ts fields as we will need them for TSDB data retrieval fields = append(fields, []string{"started_at_ts", "ended_at_ts"}...) @@ -529,22 +705,6 @@ func main() { kingpin.Fatalf("failed to parse --endtime flag: %v", err) } - // Ensure to limit period to 1 week asking for metric data - // This is to avoid fetching metrics of too many jobs when only - // period is set - if tsData && end.Sub(start) > 7*24*time.Hour { - kingpin.Fatalf("limit period between --starttime and --endtime to 7 days when --ts is enabled") - } - - // By this time, user input is validated. Time to read config file - // to get HTTP config to connect to CEEMS API server. - // Either setuid or setgid bits must be applied on the app so that - // the config file can be read as the owner of this app - config, err := readConfig(mockConfigPath) - if err != nil { - os.Exit(checkErr(fmt.Errorf("%w: %w", errConfig, err))) - } - // Setup logger promslogConfig := &promslog.Config{ Level: config.Logging.Level, @@ -623,13 +783,52 @@ func main() { logger.Info("User context changed after setuid syscall") // Get stats - units, usages, err := stats(logger, config, currentUser.Username, start, end, accounts, jobs, userNames, fields, tsData, tsDataOut) + units, usages, err := stats(logger, config, currentUser.Username, start, end, accounts, jobs, userNames, fields, summaryStats) if err != nil { os.Exit(checkErr(err)) } + // If no units found, exit, nothing more to do + if len(units) == 0 { + os.Exit(checkErr(errNoUnits)) + } + + // If instant queries have been configured, get results + var instantQueryResults map[string]map[string]string + + if len(config.TSDB.instantQueries) > 0 { + logger.Debug("Fetching instant queries results from TSDB") + + instantQueryResults, err = executeInstantQueries(logger, config, units) + if err != nil { + logger.Error("failed to fetch instant query results from TSDB", "err", err) + fmt.Fprintln(os.Stderr, "failed to fetch metrics data") + } + } + + // If tsData is enabled, get time series data + if len(config.TSDB.rangeQueries) > 0 { + // If found jobs are more than 10, print a warning + if len(units) > config.TSDB.MaxUnitsForRangeQueries { + logger.Warn("Too many jobs to fetch time series data. Ignoring --ts.metrics flag", "num_units", len(units), "max_allowed_units", config.TSDB.MaxUnitsForRangeQueries) + msg := fmt.Sprintf("too many jobs to fetch time series data. Please provide explicit job IDs (less than %d at a time) using --job when --ts.metrics is set", config.TSDB.MaxUnitsForRangeQueries) + fmt.Fprintln(os.Stderr, msg) + + goto print_table + } + + logger.Debug("Fetching time series data from TSDB") + + err := executeRangeQueries(logger, config, units, tsDataOut) + if err != nil { + logger.Error("failed to fetch time series data", "err", err) + fmt.Fprintln(os.Stderr, "failed to fetch time series data") + } + } + +print_table: // Print stats as table - t := newTable(currentUser.Username, userNames, units, usages) + t := newTable(currentUser.Username, userNames, units, usages, instantQueryResults, activeInstantQueries, summaryStats) // Based on request rendering format switch { @@ -645,7 +844,17 @@ func main() { } // newTable returns a new table with data. -func newTable(currentUser string, users []string, units []models.Unit, usages []models.Usage) table.Writer { +func newTable(currentUser string, users []string, units []models.Unit, usages []models.Usage, instantQueryResults map[string]map[string]string, activeInstantQueries []string, includeSummaryStats bool) table.Writer { + // // Get current width + // currentWidth := 180 + // // If we are in terminal override the default width with current width of terminal + // if term.IsTerminal(0) { + // width, _, err := term.GetSize(0) + // if err == nil { + // currentWidth = width + // } + // } + // Make a new writer t := table.NewWriter() @@ -659,8 +868,11 @@ func newTable(currentUser string, users []string, units []models.Unit, usages [] Color: table.ColorOptionsDefault, HTML: table.DefaultHTMLOptions, Options: table.OptionsDefault, - Size: table.SizeOptionsDefault, - Title: table.TitleOptionsDefault, + // Size: table.SizeOptions{ + // WidthMax: currentWidth, + // WidthMin: 10, + // }, + Title: table.TitleOptionsDefault, Format: table.FormatOptions{ Footer: text.FormatDefault, Header: text.FormatUpper, @@ -712,57 +924,87 @@ func newTable(currentUser string, users []string, units []models.Unit, usages [] rows := make([]table.Row, len(units)) for iunit, unit := range units { + // Marshal total time and allocation + var totalTime, allocation string + + if len(unit.TotalTime) > 0 { + val, err := json.Marshal(unit.TotalTime) + if err == nil { + totalTime = string(val) + } + } + + if len(unit.Allocation) > 0 { + val, err := json.Marshal(unit.Allocation) + if err == nil { + allocation = string(val) + } + } + row := table.Row{ unit.UUID, unit.Name, unit.Project, unit.Group, unit.User, unit.CreatedAt, - unit.StartedAt, unit.EndedAt, unit.Elapsed, unit.State, + unit.StartedAt, unit.EndedAt, unit.Elapsed, totalTime, allocation, unit.State, + } + row = append(row, unit.AveCPUUsage.Values("%.2f", len(fieldMap["cpuusage"].keys))...) + row = append(row, unit.AveCPUMemUsage.Values("%.2f", len(fieldMap["cpumemoryusage"].keys))...) + row = append(row, unit.TotalCPUEnergyUsage.Values("%f", len(fieldMap["hostenergy"].keys))...) + row = append(row, unit.TotalCPUEmissions.Values("%f", len(fieldMap["hostemissions"].keys))...) + row = append(row, unit.AveGPUUsage.Values("%.2f", len(fieldMap["gpuusage"].keys))...) + row = append(row, unit.AveGPUMemUsage.Values("%.2f", len(fieldMap["gpumemoryusage"].keys))...) + row = append(row, unit.TotalGPUEnergyUsage.Values("%f", len(fieldMap["gpuenergy"].keys))...) + row = append(row, unit.TotalGPUEmissions.Values("%f", len(fieldMap["gpuemissions"].keys))...) + + // Add instant Query results to row + for _, query := range activeInstantQueries { + row = append(row, instantQueryResults[query][unit.UUID]) } - row = append(row, unit.AveCPUUsage.Values("%.2f")...) - row = append(row, unit.AveCPUMemUsage.Values("%.2f")...) - row = append(row, unit.TotalCPUEnergyUsage.Values("%f")...) - row = append(row, unit.TotalCPUEmissions.Values("%f")...) - row = append(row, unit.AveGPUUsage.Values("%.2f")...) - row = append(row, unit.AveGPUMemUsage.Values("%.2f")...) - row = append(row, unit.TotalGPUEnergyUsage.Values("%f")...) - row = append(row, unit.TotalGPUEmissions.Values("%f")...) + rows[iunit] = row } t.AppendRows(rows) - // Append summary row - t.AppendSeparator() + if includeSummaryStats { + // Append summary row + t.AppendSeparator() - summaryRow := table.Row{"Summary"} - for range headers { - summaryRow = append(summaryRow, "") - } + summaryRow := table.Row{"Summary"} + for range headers { + summaryRow = append(summaryRow, "") + } - t.AppendRow(summaryRow, rowConfig) - t.AppendSeparator() + t.AppendRow(summaryRow, rowConfig) + t.AppendSeparator() - for _, usage := range usages { - if usage.User == currentUser || slices.Contains(users, usage.User) || slices.Contains(users, "all") { - // Check if elapsed time in non zero - var totalElapsedTime string - if usage.TotalTime["walltime"] > 0 { - totalElapsedTime = common.Timespan(time.Duration(usage.TotalTime["walltime"]) * time.Second).Format("15:04:05") - } + for _, usage := range usages { + if usage.User == currentUser || slices.Contains(users, usage.User) || slices.Contains(users, "all") { + // Check if elapsed time in non zero + var totalElapsedTime string + if usage.TotalTime["walltime"] > 0 { + totalElapsedTime = common.Timespan(time.Duration(usage.TotalTime["walltime"]) * time.Second).Format("15:04:05") + } + + // Usage row + row := table.Row{ + usage.NumUnits, "", usage.Project, usage.Group, usage.User, "", "", "", totalElapsedTime, "", "", "", + } + row = append(row, usage.AveCPUUsage.Values("%.2f", len(fieldMap["cpuusage"].keys))...) + row = append(row, usage.AveCPUMemUsage.Values("%.2f", len(fieldMap["cpumemoryusage"].keys))...) + row = append(row, usage.TotalCPUEnergyUsage.Values("%f", len(fieldMap["hostenergy"].keys))...) + row = append(row, usage.TotalCPUEmissions.Values("%f", len(fieldMap["hostemissions"].keys))...) + row = append(row, usage.AveGPUUsage.Values("%.2f", len(fieldMap["gpuusage"].keys))...) + row = append(row, usage.AveGPUMemUsage.Values("%.2f", len(fieldMap["gpumemoryusage"].keys))...) + row = append(row, usage.TotalGPUEnergyUsage.Values("%f", len(fieldMap["gpuenergy"].keys))...) + row = append(row, usage.TotalGPUEmissions.Values("%f", len(fieldMap["gpuemissions"].keys))...) + + // Append instant query results columns as "N/A" + for range activeInstantQueries { + row = append(row, "N/A") + } - // Usage row - row := table.Row{ - usage.NumUnits, "", usage.Project, usage.Group, usage.User, "", "", "", totalElapsedTime, "", + // Append row to table + t.AppendFooter(row) } - row = append(row, usage.AveCPUUsage.Values("%.2f")...) - row = append(row, usage.AveCPUMemUsage.Values("%.2f")...) - row = append(row, usage.TotalCPUEnergyUsage.Values("%f")...) - row = append(row, usage.TotalCPUEmissions.Values("%f")...) - row = append(row, usage.AveGPUUsage.Values("%.2f")...) - row = append(row, usage.AveGPUMemUsage.Values("%.2f")...) - row = append(row, usage.TotalGPUEnergyUsage.Values("%f")...) - row = append(row, usage.TotalGPUEmissions.Values("%f")...) - - // Append row to table - t.AppendFooter(row) } } @@ -783,6 +1025,9 @@ func readConfig(mockConfigPath string) (*Config, error) { var config Config // If mockConfigPath is set as well, add to configPaths + // Do not override configPaths because if there is an existing config + // sitting somewhere, we should give priority to it rather than the + // mock config if mockConfigPath != "" { configPaths = append(configPaths, mockConfigPath) } @@ -847,21 +1092,21 @@ func getCurrentUser(mockUserName string) (*user.User, error) { func parseTime(s string) (time.Time, error) { // First attempt is to parse as YYYY-MM-DDTHH:MM:SS - t, err := time.Parse("2006-01-02T15:04:05", s) + t, err := time.ParseInLocation("2006-01-02T15:04:05", s, time.Local) if err == nil { - return t.In(time.Local), nil + return t, nil } // Second attempt is to parse as YYYY-MM-DDTHH:MM - t, err = time.Parse("2006-01-02T15:04", s) + t, err = time.ParseInLocation("2006-01-02T15:04", s, time.Local) if err == nil { - return t.In(time.Local), nil + return t, nil } // Third attempt is to parse as YYYY-MM-DD - t, err = time.Parse("2006-01-02", s) + t, err = time.ParseInLocation("2006-01-02", s, time.Local) if err == nil { - return t.In(time.Local), nil + return t, nil } // If nothing works, return error @@ -905,6 +1150,8 @@ func checkErr(err error) int { fmt.Fprintln(os.Stderr, "error: "+errConfig.Error()) case errors.Is(err, errUser): fmt.Fprintln(os.Stderr, "error: "+errUser.Error()) + case errors.Is(err, errNoUnits): + fmt.Fprintln(os.Stderr, "error: "+errNoUnits.Error()) default: fmt.Fprintln(os.Stderr, "error: internal error") } diff --git a/cmd/cacct/testdata/bad-config/config.yml b/cmd/cacct/testdata/bad-config/config.yml new file mode 100644 index 00000000..9e76996b --- /dev/null +++ b/cmd/cacct/testdata/bad-config/config.yml @@ -0,0 +1,27 @@ +--- + +ceems_api_server: + cluster_id: slurm-0 + web: + url: http://localhost:9020 + + # Basic auth config + basic_auth: + username: ceems + password: password + +tsdb: + web: + url: http://localhost:9090 + max_units_for_range_queries: 2 + queries: + - name: "duplicatename" + help: "Example metric which gives vector value" + title: "Avg Metric Vector" + query: avg_over_time(avg by (uuid,instance) (avg_cpu_usage{uuid=~`{{.UUIDs}}`})[{{.Range}}:]) + kind: instant + - name: "duplicatename" + help: "Example query which returns scalar value" + title: "Avg Metric Scalar" + query: avg_over_time(avg by (uuid) (avg_cpu_usage{uuid=~`{{.UUIDs}}`})[{{.Range}}:]) + kind: instant diff --git a/cmd/cacct/testdata/config.yml b/cmd/cacct/testdata/config.yml index 987bdfe8..f1cb6c1d 100644 --- a/cmd/cacct/testdata/config.yml +++ b/cmd/cacct/testdata/config.yml @@ -13,5 +13,20 @@ ceems_api_server: tsdb: web: url: http://localhost:9090 + max_units_for_range_queries: 2 queries: - cpu_usage: avg_cpu_usage{uuid=~"%s"} + - name: cpu_usage + title: "Avg. CPU Usage" + help: "Average usage of CPU during the duration of the job" + query: avg_cpu_usage{uuid=~`{{.UUIDs}}`} + kind: range + - name: "avg_-usage-something_Vector" + help: "Example metric which gives vector value" + title: "Avg Metric Vector" + query: avg_over_time(avg by (uuid,instance) (avg_cpu_usage{uuid=~`{{.UUIDs}}`})[{{.Range}}:]) + kind: instant + - name: "Avg-UsageSomething-Scalar" + help: "Example query which returns scalar value" + title: "Avg Metric Scalar" + query: avg_over_time(avg by (uuid) (avg_cpu_usage{uuid=~`{{.UUIDs}}`})[{{.Range}}:]) + kind: instant diff --git a/cmd/cacct/testdata/output/e2e-test-cacct-bad-config.txt b/cmd/cacct/testdata/output/e2e-test-cacct-bad-config.txt new file mode 100644 index 00000000..cf31e8d3 --- /dev/null +++ b/cmd/cacct/testdata/output/e2e-test-cacct-bad-config.txt @@ -0,0 +1 @@ +error: unable to get cacct config diff --git a/cmd/cacct/testdata/output/e2e-test-cacct-custom-format.txt b/cmd/cacct/testdata/output/e2e-test-cacct-custom-format.txt index 5174ebe4..1f406208 100644 --- a/cmd/cacct/testdata/output/e2e-test-cacct-custom-format.txt +++ b/cmd/cacct/testdata/output/e2e-test-cacct-custom-format.txt @@ -1,10 +1,16 @@ -┌─────────┬─────────┬────────┐ -│ JOB ID │ ACCOUNT │ CPU US │ -│ │ │ AGE(%) │ -├─────────┼─────────┼────────┤ -│ │ │ │ -├─────────┼─────────┼────────┤ -│ 1479763 │ acc1 │ 21.22 │ -├─────────┼─────────┴────────┤ -│ Summary │ │ -└─────────┴──────────────────┘ +┌─────────┬─────────┬────────┬────────────┬────────────┐ +│ JOB ID │ ACCOUNT │ CPU US │ AVG METRIC │ AVG METRIC │ +│ │ │ AGE(%) │ VECTOR │ SCALAR │ +├─────────┼─────────┼────────┼────────────┼────────────┤ +│ │ │ │ │ │ +├─────────┼─────────┼────────┼────────────┼────────────┤ +│ 1479763 │ acc1 │ 21.22 │ [{"labels" │ [{"labels" │ +│ │ │ │ :{"instanc │ :{"instanc │ +│ │ │ │ e":"localh │ e":"localh │ +│ │ │ │ ost:9090"} │ ost:9090"} │ +│ │ │ │ ,"value":2 │ ,"value":2 │ +│ │ │ │ 1.22149394 │ 1.22149394 │ +│ │ │ │ }] │ }] │ +├─────────┼─────────┴────────┴────────────┴────────────┤ +│ Summary │ │ +└─────────┴────────────────────────────────────────────┘ diff --git a/cmd/cacct/testdata/output/e2e-test-cacct-help-format.txt b/cmd/cacct/testdata/output/e2e-test-cacct-help-format.txt new file mode 100644 index 00000000..d80d3d5b --- /dev/null +++ b/cmd/cacct/testdata/output/e2e-test-cacct-help-format.txt @@ -0,0 +1,30 @@ ++-------------------------+--------------------------------------------------------------------------+ +| FIELD | DESCRIPTION | ++-------------------------+--------------------------------------------------------------------------+ +| account | Account name | +| allocation | Resource allocation of CPU, GPU, memory, etc | +| avgUsageSomethingScalar | Example query which returns scalar value | +| avgUsageSomethingVector | Example metric which gives vector value | +| cpuMemoryUsage | Average CPU memory usage over the duration of the job | +| cpuUsage | Average CPU usage over the duration of the job | +| createdAt | Job creation time | +| elapsed | Job elapsed time | +| endedAt | Job end time | +| gpuEmissions | Total eq. emissions due to GPU(s) energy usage duration of the job | +| gpuEnergy | Total energy usage by the GPU(s) duration of the job | +| gpuMemoryUsage | Average GPU(s) memory usage over the duration of the job | +| gpuUsage | Average GPU(s) usage over the duration of the job | +| group | Group name | +| hostEmissions | Total eq. emissions due to host energy usage duration of the job | +| hostEnergy | Total energy usage by the host duration of the job | +| jobID | Job ID | +| name | Name of the job | +| startedAt | Job start time | +| state | Job state | +| totalTime | Wall, CPU and GPU times consumed in seconds over the duration of the job | +| user | User name | ++-------------------------+--------------------------------------------------------------------------+ +| Available Time Series | | ++-------------------------+--------------------------------------------------------------------------+ +| cpuUsage | Average usage of CPU during the duration of the job | ++-------------------------+--------------------------------------------------------------------------+ diff --git a/cmd/cacct/testdata/output/e2e-test-cacct-long-format.txt b/cmd/cacct/testdata/output/e2e-test-cacct-long-format.txt index 2f47a9b3..c8cdd878 100644 --- a/cmd/cacct/testdata/output/e2e-test-cacct-long-format.txt +++ b/cmd/cacct/testdata/output/e2e-test-cacct-long-format.txt @@ -1,14 +1,20 @@ -┌─────────┬──────────┬─────────┬───────┬───────┬──────────┬──────────────┬──────────────┬──────────┬───────┬────────┬────────┬──────────┬─────────────────────────┬────────┬────────┬──────────┬─────────────────────────┐ -│ JOB ID │ NAME │ ACCOUNT │ GROUP │ USER │ CREATED │ STARTED │ ENDED │ ELAPSED │ STATE │ CPU US │ CPU ME │ HOST ENE │ HOST EMISSIO │ GPU US │ GPU ME │ GPU ENER │ GPU EMISSION │ -│ │ │ │ │ │ │ │ │ │ │ AGE(%) │ M. USA │ RGY(KWH) │ NS(GMS) │ AGE(%) │ M. USA │ GY(KWH) │ S(GMS) │ -│ │ │ │ │ │ │ │ │ │ │ │ GE(%) │ │ │ │ GE(%) │ │ │ -├─────────┼──────────┼─────────┼───────┼───────┼──────────┼──────────────┼──────────────┼──────────┼───────┼────────┼────────┼──────────┼─────────────┬───────────┼────────┼────────┼──────────┼─────────────┬───────────┤ -│ │ │ │ │ │ │ │ │ │ │ │ │ │ EMAPS_TOTAL │ RTE_TOTAL │ │ │ │ EMAPS_TOTAL │ RTE_TOTAL │ -├─────────┼──────────┼─────────┼───────┼───────┼──────────┼──────────────┼──────────────┼──────────┼───────┼────────┼────────┼──────────┼─────────────┼───────────┼────────┼────────┼──────────┼─────────────┼───────────┤ -│ 1479763 │ test_scr │ acc1 │ grp1 │ usr1 │ 2022-02- │ 2022-02-21T1 │ 2022-02-21T1 │ 00:49:22 │ CANCE │ 21.22 │ 21.22 │ 21.22149 │ 21.221494 │ 21.221494 │ 21.22 │ 21.22 │ 21.22149 │ 21.221494 │ 21.221494 │ -│ │ ipt1 │ │ │ │ 21T14:37 │ 4:37:07+0100 │ 5:26:29+0100 │ │ LLED │ │ │ 4 │ │ │ │ │ 4 │ │ │ -│ │ │ │ │ │ :02+0100 │ │ │ │ by 10 │ │ │ │ │ │ │ │ │ │ │ -│ │ │ │ │ │ │ │ │ │ 01 │ │ │ │ │ │ │ │ │ │ │ -├─────────┼──────────┴─────────┴───────┴───────┴──────────┴──────────────┴──────────────┴──────────┴───────┴────────┴────────┴──────────┴─────────────┴───────────┴────────┴────────┴──────────┴─────────────┴───────────┤ -│ Summary │ │ -└─────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ +┌─────────┬──────────┬─────────┬───────┬───────┬──────────┬──────────────┬──────────────┬──────────┬──────────────┬──────────────┬───────┬────────┬────────┬──────────┬─────────────────────────┬────────┬────────┬──────────┬─────────────────────────┬────────────┬────────────┐ +│ JOB ID │ NAME │ ACCOUNT │ GROUP │ USER │ CREATED │ STARTED │ ENDED │ ELAPSED │ TOTAL TIME(S │ RESOURCE ALL │ STATE │ CPU US │ CPU ME │ HOST ENE │ HOST EMISSIO │ GPU US │ GPU ME │ GPU ENER │ GPU EMISSION │ AVG METRIC │ AVG METRIC │ +│ │ │ │ │ │ │ │ │ │ ) │ OCATION │ │ AGE(%) │ M. USA │ RGY(KWH) │ NS(GMS) │ AGE(%) │ M. USA │ GY(KWH) │ S(GMS) │ VECTOR │ SCALAR │ +│ │ │ │ │ │ │ │ │ │ │ │ │ │ GE(%) │ │ │ │ GE(%) │ │ │ │ │ +├─────────┼──────────┼─────────┼───────┼───────┼──────────┼──────────────┼──────────────┼──────────┼──────────────┼──────────────┼───────┼────────┼────────┼──────────┼─────────────┬───────────┼────────┼────────┼──────────┼─────────────┬───────────┼────────────┼────────────┤ +│ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ EMAPS_TOTAL │ RTE_TOTAL │ │ │ │ EMAPS_TOTAL │ RTE_TOTAL │ │ │ +├─────────┼──────────┼─────────┼───────┼───────┼──────────┼──────────────┼──────────────┼──────────┼──────────────┼──────────────┼───────┼────────┼────────┼──────────┼─────────────┼───────────┼────────┼────────┼──────────┼─────────────┼───────────┼────────────┼────────────┤ +│ 1479763 │ test_scr │ acc1 │ grp1 │ usr1 │ 2022-02- │ 2022-02-21T1 │ 2022-02-21T1 │ 00:49:22 │ {"alloc_cpum │ {"billing":8 │ CANCE │ 21.22 │ 21.22 │ 21.22149 │ 21.221494 │ 21.221494 │ 21.22 │ 21.22 │ 21.22149 │ 21.221494 │ 21.221494 │ [{"labels" │ [{"labels" │ +│ │ ipt1 │ │ │ │ 21T14:37 │ 4:37:07+0100 │ 5:26:29+0100 │ │ emtime":9705 │ 0,"cpus":8," │ LLED │ │ │ 4 │ │ │ │ │ 4 │ │ │ :{"instanc │ :{"instanc │ +│ │ │ │ │ │ :02+0100 │ │ │ │ 88160,"alloc │ gpus":8,"mem │ by 10 │ │ │ │ │ │ │ │ │ │ │ e":"localh │ e":"localh │ +│ │ │ │ │ │ │ │ │ │ _cputime":23 │ ":3435973836 │ 01 │ │ │ │ │ │ │ │ │ │ │ ost:9090"} │ ost:9090"} │ +│ │ │ │ │ │ │ │ │ │ 696,"alloc_g │ 80,"nodes":1 │ │ │ │ │ │ │ │ │ │ │ │ ,"value":2 │ ,"value":2 │ +│ │ │ │ │ │ │ │ │ │ pumemtime":2 │ } │ │ │ │ │ │ │ │ │ │ │ │ 1.22149394 │ 1.22149394 │ +│ │ │ │ │ │ │ │ │ │ 962,"alloc_g │ │ │ │ │ │ │ │ │ │ │ │ │ }] │ }] │ +│ │ │ │ │ │ │ │ │ │ putime":2369 │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ +│ │ │ │ │ │ │ │ │ │ 6,"walltime" │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ +│ │ │ │ │ │ │ │ │ │ :2962} │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ +├─────────┼──────────┴─────────┴───────┴───────┴──────────┴──────────────┴──────────────┴──────────┴──────────────┴──────────────┴───────┴────────┴────────┴──────────┴─────────────┴───────────┴────────┴────────┴──────────┴─────────────┴───────────┴────────────┴────────────┤ +│ Summary │ │ +└─────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ diff --git a/cmd/cacct/testdata/output/e2e-test-cacct-tsdata-fail.txt b/cmd/cacct/testdata/output/e2e-test-cacct-tsdata-fail.txt index 1e7007e4..b41a6c2b 100644 --- a/cmd/cacct/testdata/output/e2e-test-cacct-tsdata-fail.txt +++ b/cmd/cacct/testdata/output/e2e-test-cacct-tsdata-fail.txt @@ -1 +1,30 @@ -cacct: error: limit period between --starttime and --endtime to 7 days when --ts is enabled +too many jobs to fetch time series data. Please provide explicit job IDs (less than 2 at a time) using --job when --ts.metrics is set +┌─────────┬─────────┬───────┬──────────┬────────┬────────┬──────────┬─────────────────────────┬────────┬────────┬──────────┬─────────────────────────┐ +│ JOB ID │ ACCOUNT │ USER │ ELAPSED │ CPU US │ CPU ME │ HOST ENE │ HOST EMISSIO │ GPU US │ GPU ME │ GPU ENER │ GPU EMISSION │ +│ │ │ │ │ AGE(%) │ M. USA │ RGY(KWH) │ NS(GMS) │ AGE(%) │ M. USA │ GY(KWH) │ S(GMS) │ +│ │ │ │ │ │ GE(%) │ │ │ │ GE(%) │ │ │ +│ │ │ │ │ │ │ │ EMAPS_TOTAL │ RTE_TOTAL │ │ │ │ EMAPS_TOTAL │ RTE_TOTAL │ +├─────────┼─────────┼───────┼──────────┼────────┼────────┼──────────┼─────────────┼───────────┼────────┼────────┼──────────┼─────────────┼───────────┤ +│ 1009248 │ testacc │ testu │ 00:00:17 │ 21.23 │ 21.23 │ 21.23464 │ 21.234647 │ 21.234647 │ 21.23 │ 21.23 │ 21.23464 │ 21.234647 │ 21.234647 │ +│ │ │ sr │ │ │ │ 7 │ │ │ │ │ 7 │ │ │ +│ 11508 │ acc1 │ usr15 │ 00:08:17 │ 17.80 │ 17.80 │ 17.79969 │ 17.799693 │ 17.799693 │ 17.80 │ 17.80 │ 17.79969 │ 17.799693 │ 17.799693 │ +│ │ │ │ │ │ │ 3 │ │ │ │ │ 3 │ │ │ +│ 14508 │ acc4 │ usr4 │ 00:08:17 │ 14.03 │ 14.03 │ 14.03205 │ 14.032058 │ 14.032058 │ 14.03 │ 14.03 │ 14.03205 │ 14.032058 │ 14.032058 │ +│ │ │ │ │ │ │ 8 │ │ │ │ │ 8 │ │ │ +│ 147973 │ acc2 │ usr1 │ 00:00:17 │ 29.39 │ 29.39 │ 29.38529 │ 29.385290 │ 29.385290 │ 29.39 │ 29.39 │ 29.38529 │ 29.385290 │ 29.385290 │ +│ │ │ │ │ │ │ 0 │ │ │ │ │ 0 │ │ │ +│ 147975 │ acc3 │ usr3 │ 00:49:22 │ 29.72 │ 29.72 │ 29.72084 │ 29.720843 │ 29.720843 │ 29.72 │ 29.72 │ 29.72084 │ 29.720843 │ 29.720843 │ +│ │ │ │ │ │ │ 3 │ │ │ │ │ 3 │ │ │ +│ 1479763 │ acc1 │ usr1 │ 00:49:22 │ 21.22 │ 21.22 │ 21.22149 │ 21.221494 │ 21.221494 │ 21.22 │ 21.22 │ 21.22149 │ 21.221494 │ 21.221494 │ +│ │ │ │ │ │ │ 4 │ │ │ │ │ 4 │ │ │ +│ 1479765 │ acc1 │ usr8 │ 00:49:22 │ 20.21 │ 20.21 │ 20.21483 │ 20.214837 │ 20.214837 │ 20.21 │ 20.21 │ 20.21483 │ 20.214837 │ 20.214837 │ +│ │ │ │ │ │ │ 7 │ │ │ │ │ 7 │ │ │ +│ 1481508 │ acc2 │ usr2 │ 00:08:17 │ 53.48 │ 53.48 │ 53.47701 │ 53.477015 │ 53.477015 │ 53.48 │ 53.48 │ 53.47701 │ 53.477015 │ 53.477015 │ +│ │ │ │ │ │ │ 5 │ │ │ │ │ 5 │ │ │ +│ 1481510 │ acc3 │ usr3 │ 00:00:17 │ 50.14 │ 50.14 │ 50.13620 │ 50.136201 │ 50.136201 │ 50.14 │ 50.14 │ 50.13620 │ 50.136201 │ 50.136201 │ +│ │ │ │ │ │ │ 1 │ │ │ │ │ 1 │ │ │ +│ 81510 │ acc1 │ usr15 │ 00:00:17 │ 18.57 │ 18.57 │ 18.57046 │ 18.570466 │ 18.570466 │ 18.57 │ 18.57 │ 18.57046 │ 18.570466 │ 18.570466 │ +│ │ │ │ │ │ │ 6 │ │ │ │ │ 6 │ │ │ +├─────────┼─────────┴───────┴──────────┴────────┴────────┴──────────┴─────────────┴───────────┴────────┴────────┴──────────┴─────────────┴───────────┤ +│ Summary │ │ +└─────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ diff --git a/cmd/cacct/testdata/output/e2e-test-cacct-tsdata.txt b/cmd/cacct/testdata/output/e2e-test-cacct-tsdata.txt index b99b4f3d..d1a6c40b 100644 --- a/cmd/cacct/testdata/output/e2e-test-cacct-tsdata.txt +++ b/cmd/cacct/testdata/output/e2e-test-cacct-tsdata.txt @@ -20,8 +20,9 @@ { "fingerprint": "554b56cadf9dea4b", "labels": { - "__name__": "cpu_usage", - "instance": "localhost:9090", + "help": "Average usage of CPU during the duration of the job", + "metric": "Avg. CPU Usage", + "nodename": "localhost", "uuid": "147973" } } diff --git a/cmd/cacct/tsdb.go b/cmd/cacct/tsdb.go index 4b6747eb..f066207d 100644 --- a/cmd/cacct/tsdb.go +++ b/cmd/cacct/tsdb.go @@ -2,24 +2,31 @@ package main import ( "bytes" + "cmp" "context" "encoding/csv" "encoding/json" + "errors" "fmt" + "html/template" "log/slog" "os" "path/filepath" + "slices" + "strings" "sync" "time" + "github.com/ceems-dev/ceems/internal/common" "github.com/ceems-dev/ceems/pkg/api/models" "github.com/ceems-dev/ceems/pkg/tsdb" "github.com/prometheus/common/model" ) var ( - queryMDMu = sync.RWMutex{} - queryMD []queryMetadata + queryMDMu = sync.RWMutex{} + queryInstant = sync.RWMutex{} + queryMD []queryMetadata ) // queryMetadata contains metadata information for each TSDB series. We dump @@ -32,10 +39,27 @@ type queryMetadata struct { Labels model.Metric `json:"labels"` } -// tsdbData saves time series data of units in CSV files. -func tsdbData(ctx context.Context, logger *slog.Logger, config *Config, units []models.Unit, outDir string) error { +// CacctSample is a custom sample that we use in exporting data in CSV, JSON formats. +type CacctSample struct { + Labels map[string]string `json:"labels"` + Value float64 `json:"value"` +} + +// unitsBatch is a container to keep slice of UUIDs of each batch. +type unitsBatch struct { + uuids []string + evaluationInterval time.Duration + duration time.Duration + queryTime int64 +} + +// executeRangeQueries executes range queries and saves results to CSV files. +func executeRangeQueries(logger *slog.Logger, config *Config, units []models.Unit, outDir string) error { + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(config.TSDB.QueryTimeout)) + defer cancel() + // New TSDB client - tsdb, err := tsdb.New(config.TSDB.Web.URL, config.TSDB.Web.HTTPClientConfig, slog.New(slog.DiscardHandler)) + client, err := tsdb.New(config.TSDB.Web.URL, config.TSDB.Web.HTTPClientConfig, slog.New(slog.DiscardHandler)) if err != nil { logger.Error("Failed to create a new TSDB client", "err", err) @@ -56,38 +80,69 @@ func tsdbData(ctx context.Context, logger *slog.Logger, config *Config, units [] logger.Debug("Time series data will be saved", "dir", absOutDir) + // Get TSDB settings + settings := getTSDBSettings(ctx, config, client) + // Start a wait group wg := sync.WaitGroup{} // Fetch time series of each metric in separate go routine for _, unit := range units { - for queryID, query := range config.TSDB.Queries { + for _, q := range config.TSDB.rangeQueries { wg.Add(1) + // Template data + // LSF job arrays will have IDs like 300[1], 300[2], etc. Prometheus expects the + // square brackets to be escaped or else it will ignore the label values. This is + // due to the fact that it will use regex expression to match the label values. + tmplData := map[string]any{ + "UUIDs": strings.ReplaceAll(strings.ReplaceAll(unit.UUID, "[", `\[`), "]", `\]`), + "ScrapeInterval": settings.ScrapeInterval, + "ScrapeIntervalMilli": settings.ScrapeInterval.Milliseconds(), + "EvaluationInterval": settings.EvaluationInterval, + "EvaluationIntervalMilli": settings.EvaluationInterval.Milliseconds(), + "RateInterval": settings.RateInterval, + "Range": time.Duration((unit.EndedAtTS - unit.StartedAtTS) * int64(time.Millisecond)), + } + + // Build query + query, err := queryBuilder(q.Name, q.Query, tmplData) + if err != nil { + logger.Error("Failed to build TSDB query", "query", q.Query, "err", err) + wg.Done() + + continue + } + // Fetch metrics from TSDB and write to CSV files - go fetchData(ctx, queryID, fmt.Sprintf(query, unit.UUID), unit.StartedAtTS, unit.EndedAtTS, absOutDir, tsdb, &wg) + go fetchRangeData(ctx, logger, q, query, unit.StartedAtTS, unit.EndedAtTS, absOutDir, client, &wg) } } // Wait for all routines wg.Wait() - // Dump metadata.json - writeMetadata(logger, queryMD, absOutDir) + // Dump metadata.json for time series data + if len(queryMD) > 0 { + writeMetadata(logger, queryMD, absOutDir) - fmt.Fprintln(os.Stderr, "time series data saved to directory", absOutDir) + fmt.Fprintln(os.Stderr, "time series data saved to directory", absOutDir) + } else { + return errors.New("no metadata found for range queries") + } return nil } -// fetchData retrieves time series data from TSDB. -func fetchData(ctx context.Context, queryID string, query string, start int64, end int64, outDir string, tsdb *tsdb.Client, wg *sync.WaitGroup) { +// fetchRangeData retrieves range query results from TSDB. +func fetchRangeData(ctx context.Context, logger *slog.Logger, q TSDBQuery, query string, start int64, end int64, outDir string, client *tsdb.Client, wg *sync.WaitGroup) { defer wg.Done() // Make a range query - results, err := tsdb.RangeQuery(ctx, query, time.UnixMilli(start), time.UnixMilli(end), 10*time.Second, time.Minute) + results, err := client.RangeQuery(ctx, query, time.UnixMilli(start), time.UnixMilli(end), 10*time.Second, time.Minute) if err != nil { - fmt.Fprintln(os.Stderr, "failed to fetch time series for query", query, "err:", err) + logger.Error("Failed to fetch time series for query", "query", query, "err", err) + fmt.Fprintln(os.Stderr, "failed to fetch time series query", q.Name) return } @@ -106,7 +161,26 @@ func fetchData(ctx context.Context, queryID string, query string, start int64, e // Replace series name in labels with queryID // This is more readable one and also allows us // to protect Prometheus series names - labels["__name__"] = model.LabelValue(queryID) + labels["metric"] = model.LabelValue(q.Name) + if q.Title != "" { + labels["metric"] = model.LabelValue(q.Title) + } + + if q.Help != "" { + labels["help"] = model.LabelValue(q.Help) + } + + delete(labels, "__name__") + + // Strip port number, if exists, from instance and rename it to nodename + labels["nodename"] = model.LabelValue(strings.Split(string(labels["instance"]), ":")[0]) + delete(labels, "instance") + + // Delete Prometheus specific labels + delete(labels, "job") + delete(labels, "hostname") + delete(labels, "manager") + delete(labels, "cgrouphostname") // Add metadata of query md = append(md, queryMetadata{ @@ -156,6 +230,150 @@ func fetchData(ctx context.Context, queryID string, query string, start int64, e queryMD = append(queryMD, md...) } +// executeInstantQueries executes instant queries and returns map of query results. +func executeInstantQueries(logger *slog.Logger, config *Config, units []models.Unit) (map[string]map[string]string, error) { + // New TSDB client + client, err := tsdb.New(config.TSDB.Web.URL, config.TSDB.Web.HTTPClientConfig, slog.New(slog.DiscardHandler)) + if err != nil { + logger.Error("Failed to create a new TSDB client", "err", err) + + return nil, fmt.Errorf("failed to create tsdb API client: %w", err) + } + + // Context for settings query + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + + // Get TSDB settings + settings := getTSDBSettings(ctx, config, client) + + // If scrape and evaluation intervals have been provided, use them instead of global value + if config.TSDB.ScrapeInterval > 0 { + settings.ScrapeInterval = time.Duration(config.TSDB.ScrapeInterval) + settings.RateInterval = 4 * time.Duration(config.TSDB.ScrapeInterval) + } + + if config.TSDB.EvaluationInterval > 0 { + settings.EvaluationInterval = time.Duration(config.TSDB.EvaluationInterval) + } + + // Chunk units by their duration to make optimized queries to TSDB by avoiding + // loading too many samples in the memory + unitBatches := chunkByDuration(units, settings, config) + + allInstantResults := make(map[string]map[string]string) + + // Initialise inner maps in allInstantResults + for _, q := range config.TSDB.instantQueries { + allInstantResults[q.Name] = make(map[string]string) + } + + // Start a new context for making actual queries + ctx, cancel = context.WithTimeout(context.Background(), time.Duration(config.TSDB.QueryTimeout)) + defer cancel() + + // Fetch instant query results + numBatches := len(unitBatches) + for iBatch, unitBatch := range unitBatches { + batchSize := len(unitBatch.uuids) + // If batch is empty, skip + if batchSize == 0 { + continue + } + + // Start a wait group for each batch + wg := sync.WaitGroup{} + for _, q := range config.TSDB.instantQueries { + wg.Add(1) + + // Template data + // LSF job arrays will have IDs like 300[1], 300[2], etc. Prometheus expects the + // square brackets to be escaped or else it will ignore the label values. This is + // due to the fact that it will use regex expression to match the label values. + tmplData := map[string]any{ + "UUIDs": strings.ReplaceAll(strings.ReplaceAll(strings.Join(unitBatch.uuids, "|"), "[", `\[`), "]", `\]`), + "ScrapeInterval": settings.ScrapeInterval, + "ScrapeIntervalMilli": settings.ScrapeInterval.Milliseconds(), + "EvaluationInterval": unitBatch.evaluationInterval, + "EvaluationIntervalMilli": unitBatch.evaluationInterval.Milliseconds(), + "RateInterval": settings.RateInterval, + "Range": unitBatch.duration, + } + + // Build query + query, err := queryBuilder(q.Name, q.Query, tmplData) + if err != nil { + logger.Error("Failed to build TSDB query", "query", q.Query, "err", err) + wg.Done() + + continue + } + + // Fetch instant query metrics from TSDB + go fetchInstantData(ctx, logger, q, query, time.UnixMilli(unitBatch.queryTime), allInstantResults, client, &wg) + } + + // Wait for all routines + wg.Wait() + + logger.Debug( + "Instant queries execution progress", "batch_id", iBatch, "total_batches", numBatches, "batch_size", batchSize, "batch_duration", unitBatch.duration, + "batch_query_time", unitBatch.queryTime, "batch_evaluation_interval", unitBatch.evaluationInterval, + ) + } + + return allInstantResults, nil +} + +// fetchInstantData retrieves results of instant queries from TSDB. +func fetchInstantData(ctx context.Context, logger *slog.Logger, q TSDBQuery, query string, queryTime time.Time, allResults map[string]map[string]string, client *tsdb.Client, wg *sync.WaitGroup) { + defer wg.Done() + + // Make a instant query + results, err := client.Query(ctx, query, queryTime, time.Minute) + if err != nil { + logger.Error("Failed to fetch instant query results", "query", query, "err", err) + + return + } + // Append all the results to allResults maps + queryResults := make(map[string][]CacctSample) + + for _, sample := range results { + var ( + uuid string + entry CacctSample + ) + + // Intialise CacctSample.Labels maps + entry.Labels = make(map[string]string) + + for ln, lv := range sample.Metric { + if string(ln) == "uuid" { + uuid = string(lv) + + continue + } + + entry.Labels[string(ln)] = string(lv) + } + + entry.Value = float64(sample.Value) + queryResults[uuid] = append(queryResults[uuid], entry) + } + + // Append current results to all results + queryInstant.Lock() + defer queryInstant.Unlock() + + for uuid, values := range queryResults { + jsonString, err := json.Marshal(values) + if err == nil { + allResults[q.Name][uuid] = string(jsonString) + } + } +} + // writeMetadata dumps the metadata.json file to outDir. func writeMetadata(logger *slog.Logger, mds []queryMetadata, outDir string) { metadataFilepath := filepath.Join(outDir, "metadata.json") @@ -202,6 +420,107 @@ func writeMetadata(logger *slog.Logger, mds []queryMetadata, outDir string) { logger.Debug("Metadata file saved", "file", metadataFilepath) } +// getTSDBSettings return TSDB settings after overriding intervals from provided config. +func getTSDBSettings(ctx context.Context, config *Config, client *tsdb.Client) *tsdb.Settings { + // Get current TSDB settings + // Get rate and scrape intervals + settings := client.Settings(ctx) + + // If scrape and evaluation intervals have been provided, use them instead of global value + if config.TSDB.ScrapeInterval > 0 { + settings.ScrapeInterval = time.Duration(config.TSDB.ScrapeInterval) + settings.RateInterval = 4 * time.Duration(config.TSDB.ScrapeInterval) + } + + if config.TSDB.EvaluationInterval > 0 { + settings.EvaluationInterval = time.Duration(config.TSDB.EvaluationInterval) + } + + return settings +} + +// chunkByDuration chunks units slice into sub slices based on the duration of each unit. The idea +// is to estimate number of samples required for each chunk and keep that sum less than +// TSDB's query max-samples value. +func chunkByDuration(units []models.Unit, settings *tsdb.Settings, config *Config) []unitsBatch { + // Find the latest timestamp of unit termination. In the worst case scenario we + // will evaluate all queries until this time. So, this should give us the maximum + // duration + var maxEndedAtTS int64 + for iunit := range units { + if units[iunit].EndedAtTS > maxEndedAtTS { + maxEndedAtTS = units[iunit].EndedAtTS + } + } + + // Sort units by walltime so TSDB will have better cache efficiency when making queries + numUnits := len(units) + + unitsTmp := make([]models.Unit, numUnits) + copy(unitsTmp, units) + + // Sort unitsTmp in asc order based on walltime relative to latest unit end time + slices.SortFunc(unitsTmp, func(a, b models.Unit) int { + return cmp.Compare(maxEndedAtTS-a.StartedAtTS, maxEndedAtTS-b.StartedAtTS) + }) + + // Available samples count + availableSamples := int64(float64(settings.QueryMaxSamples) * config.TSDB.QueryMinSamples) + + // Get samples of each unit + // Here we estimate the samples assuming the unit ends at the latest timestamp amongst + // all the units fetched + var unitSamples []int64 + for iunit := range unitsTmp { + unitSamples = append(unitSamples, max((maxEndedAtTS-unitsTmp[iunit].StartedAtTS)/settings.ScrapeInterval.Milliseconds(), 1)*config.TSDB.QueryMaxSeries) + } + + // Chunk unitSamples into chunks where sum of each chunk do not exceed availableSamples + chunks := common.ChunkByMaxSum(unitSamples, availableSamples) + + unitBatches := make([]unitsBatch, len(chunks)) + + iunit := 0 + for ichunk, chunk := range chunks { + minStartedTS := unitsTmp[iunit].StartedAtTS + maxEndedTS := unitsTmp[iunit].EndedAtTS + chunkSamples := int64(0) + + for _, numSamples := range chunk { + unitBatches[ichunk].uuids = append(unitBatches[ichunk].uuids, unitsTmp[iunit].UUID) + if unitsTmp[iunit].StartedAtTS < minStartedTS { + minStartedTS = unitsTmp[iunit].StartedAtTS + } + + if unitsTmp[iunit].EndedAtTS > maxEndedTS { + maxEndedTS = unitsTmp[iunit].EndedAtTS + } + + iunit++ + chunkSamples += numSamples + } + + unitBatches[ichunk].duration = time.Duration((maxEndedTS - minStartedTS) * int64(time.Millisecond)) + unitBatches[ichunk].queryTime = maxEndedTS + unitBatches[ichunk].evaluationInterval = time.Duration(settings.EvaluationInterval.Seconds()*max(float64(chunkSamples)/float64(availableSamples), 1.0)) * time.Second + } + + return unitBatches +} + +// queryBuilder builds query from template and data. +func queryBuilder(name string, queryTemplate string, data map[string]any) (string, error) { + tmpl := template.Must(template.New(name).Parse(queryTemplate)) + builder := &strings.Builder{} + + err := tmpl.Execute(builder, data) + if err != nil { + return "", err + } + + return builder.String(), nil +} + // newCSVWriter returns a new CSV writer. func newCSVWriter(filename string) (*csv.Writer, *os.File, error) { f, err := os.Create(filename) diff --git a/cmd/redfish_proxy/reverseproxy.go b/cmd/redfish_proxy/reverseproxy.go index cd322bc1..f242ba12 100644 --- a/cmd/redfish_proxy/reverseproxy.go +++ b/cmd/redfish_proxy/reverseproxy.go @@ -48,7 +48,7 @@ func NewMultiHostReverseProxy(c *rpConfig) (*httputil.ReverseProxy, error) { return nil, err } - director := func(req *http.Request) { + rewrite := func(req *httputil.ProxyRequest) { rewriteRequestURL(c.logger, req, targets) } @@ -60,7 +60,7 @@ func NewMultiHostReverseProxy(c *rpConfig) (*httputil.ReverseProxy, error) { rw.Write([]byte("failed to find redfish target")) } - return &httputil.ReverseProxy{Director: director, Transport: httpRoundTripper, ErrorHandler: errorHandler}, nil + return &httputil.ReverseProxy{Rewrite: rewrite, Transport: httpRoundTripper, ErrorHandler: errorHandler}, nil } // rewriteRequestURL rewrites the request URL to point to the target. @@ -72,7 +72,7 @@ func NewMultiHostReverseProxy(c *rpConfig) (*httputil.ReverseProxy, error) { // // Always X-Redfish-Url header is checked for BMC hostname and if not found, // target URL is looked up from provided targets. -func rewriteRequestURL(logger *slog.Logger, req *http.Request, targets map[string]*url.URL) { +func rewriteRequestURL(logger *slog.Logger, preq *httputil.ProxyRequest, targets map[string]*url.URL) { var target *url.URL var remoteIPs []string @@ -82,10 +82,10 @@ func rewriteRequestURL(logger *slog.Logger, req *http.Request, targets map[strin var ok bool // First get the remote address of the client - remoteIPs = req.Header[http.CanonicalHeaderKey(realIPHeaderName)] + remoteIPs = preq.In.Header[http.CanonicalHeaderKey(realIPHeaderName)] // Add remoteAddr only when not on testing - ip, _, err := net.SplitHostPort(req.RemoteAddr) + ip, _, err := net.SplitHostPort(preq.In.RemoteAddr) if err == nil && os.Getenv("__IS_TESTING") == "" { remoteIPs = append(remoteIPs, ip) } @@ -107,7 +107,7 @@ func rewriteRequestURL(logger *slog.Logger, req *http.Request, targets map[strin // If target is not found in map, check header // Always use CanonicalHeaderKey as golang always canonicalize headers // internally - if targetURL := req.Header.Get(redfishURLHeaderName); targetURL != "" { + if targetURL := preq.In.Header.Get(redfishURLHeaderName); targetURL != "" { target, err = url.Parse(targetURL) if err != nil { logger.Error("Fetched Redfish URL from headers is invalid", "err", err) @@ -136,21 +136,21 @@ rewrite_req: targetQuery := target.RawQuery - req.URL.Scheme = target.Scheme - req.URL.Host = target.Host - req.URL.Path, req.URL.RawPath = joinURLPath(target, req.URL) + preq.Out.URL.Scheme = target.Scheme + preq.Out.URL.Host = target.Host + preq.Out.URL.Path, preq.Out.URL.RawPath = joinURLPath(target, preq.Out.URL) - if targetQuery == "" || req.URL.RawQuery == "" { - req.URL.RawQuery = targetQuery + req.URL.RawQuery + if targetQuery == "" || preq.Out.URL.RawQuery == "" { + preq.Out.URL.RawQuery = targetQuery + preq.Out.URL.RawQuery } else { - req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery + preq.Out.URL.RawQuery = targetQuery + "&" + preq.Out.URL.RawQuery } // Strip X-Redfish-Url header before proxying request to target - req.Header.Del(redfishURLHeaderName) + preq.Out.Header.Del(redfishURLHeaderName) // Strip Authorization header as well - req.Header.Del(authorization) + preq.Out.Header.Del(authorization) } func singleJoiningSlash(a, b string) string { diff --git a/examples/mock_resource_manager/pkg/resource/mock_manager.go b/examples/mock_resource_manager/pkg/resource/mock_manager.go index 60c5028c..2dbf1592 100644 --- a/examples/mock_resource_manager/pkg/resource/mock_manager.go +++ b/examples/mock_resource_manager/pkg/resource/mock_manager.go @@ -77,33 +77,6 @@ func (s *mockManager) FetchUnits(_ context.Context, _ time.Time, _ time.Time) ([ } // FetchUsersProjects returnc current users and projects. -func (s *mockManager) FetchUsersProjects( - _ context.Context, - _ time.Time, -) ([]models.ClusterUsers, []models.ClusterProjects, error) { - return []models.ClusterUsers{ - { - Cluster: models.Cluster{ - ID: "mock", - }, - Users: []models.User{ - { - Name: "usr1", - Projects: models.List{"prj1", "prj2"}, - }, - }, - }, - }, []models.ClusterProjects{ - { - Cluster: models.Cluster{ - ID: "mock", - }, - Projects: []models.Project{ - { - Name: "usr1", - Users: models.List{"prj1", "prj2"}, - }, - }, - }, - }, nil +func (s *mockManager) FetchUsersProjects(_ context.Context, _ time.Time) ([]models.ClusterUsers, []models.ClusterProjects, error) { + return []models.ClusterUsers{{Cluster: models.Cluster{ID: "mock"}, Users: []models.User{{Name: "usr1", Projects: models.List{"prj1", "prj2"}}}}}, []models.ClusterProjects{{Cluster: models.Cluster{ID: "mock"}, Projects: []models.Project{{Name: "usr1", Users: models.List{"prj1", "prj2"}}}}}, nil } diff --git a/go.mod b/go.mod index bfd324c6..806bac98 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,7 @@ require ( github.com/gorilla/mux v1.8.1 github.com/grafana/pyroscope/api v1.5.0 github.com/grafana/pyroscope/ebpf v0.4.11 + github.com/iancoleman/strcase v0.3.0 github.com/jedib0t/go-pretty/v6 v6.8.3 github.com/jellydator/ttlcache/v3 v3.4.1 github.com/mattn/go-sqlite3 v1.14.49 diff --git a/go.sum b/go.sum index 3cab9ea5..8a7d42cc 100644 --- a/go.sum +++ b/go.sum @@ -117,6 +117,8 @@ github.com/grafana/regexp v0.0.0-20250905093917-f7b3be9d1853 h1:cLN4IBkmkYZNnk7E github.com/grafana/regexp v0.0.0-20250905093917-f7b3be9d1853/go.mod h1:+JKpmjMGhpgPL+rXZ5nsZieVzvarn86asRlBg4uNGnk= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/iancoleman/strcase v0.3.0 h1:nTXanmYxhfFAMjZL34Ov6gkzEsSJZ5DbhxWjvSASxEI= +github.com/iancoleman/strcase v0.3.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho= github.com/ianlancetaylor/demangle v0.0.0-20251118225945-96ee0021ea0f h1:Fnl4pzx8SR7k7JuzyW8lEtSFH6EQ8xgcypgIn8pcGIE= github.com/ianlancetaylor/demangle v0.0.0-20251118225945-96ee0021ea0f/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw= github.com/jedib0t/go-pretty/v6 v6.8.3 h1:yVSk5aemoYHCvcrtqyXklwqcgHQIQzmy/oUzFlmffSQ= diff --git a/internal/common/helpers.go b/internal/common/helpers.go index c069e8e5..57d41c2c 100644 --- a/internal/common/helpers.go +++ b/internal/common/helpers.go @@ -515,3 +515,61 @@ func CheckHTTPClientConfigFiles(config *config_util.HTTPClientConfig) ([]string, return readPaths, nil } + +// ChunkByMaxSum chunks the slice into chunks so that sum of each chunk will not exceed +// maxSum. +func ChunkByMaxSum[T int | int16 | int32 | int64 | uint | uint16 | uint32 | uint64](nums []T, maxSum T) [][]T { + var result [][]T + if maxSum <= 0 { + return result + } + + var ( + currentChunk []T + currentSum T = 0 + ) + + for _, num := range nums { + if currentSum+num > maxSum { + if len(currentChunk) > 0 { + result = append(result, currentChunk) + } + + currentChunk = []T{num} + currentSum = num + } else { + currentChunk = append(currentChunk, num) + currentSum += num + } + } + + if len(currentChunk) > 0 { + result = append(result, currentChunk) + } + + return result +} + +// TimeToTimestamp converts a date in a given layout to unix timestamp of the date. +func TimeToTimestamp(layout string, date string) int64 { + t, err := time.Parse(layout, date) + if err == nil { + return t.UnixMilli() + } + + return 0 +} + +// ChunkBy splits the slice into chunks of given size. +func ChunkBy[T any](items []T, chunkSize int) [][]T { + if chunkSize == 0 { + return [][]T{items} + } + + _chunks := make([][]T, 0, (len(items)/chunkSize)+1) + for chunkSize < len(items) { + items, _chunks = items[chunkSize:], append(_chunks, items[0:chunkSize:chunkSize]) + } + + return append(_chunks, items) +} diff --git a/internal/common/helpers_test.go b/internal/common/helpers_test.go index 8fe39a4c..9f74389f 100644 --- a/internal/common/helpers_test.go +++ b/internal/common/helpers_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "github.com/ceems-dev/ceems/pkg/api/base" "github.com/ceems-dev/ceems/pkg/grafana" "github.com/prometheus/common/config" "github.com/stretchr/testify/assert" @@ -660,3 +661,93 @@ func TestCheckHTTPClientConfigFiles(t *testing.T) { assert.ElementsMatch(t, test.expected, got, test.name) } } + +func TestChunkByMaxSum(t *testing.T) { + tests := []struct { + name string + input []int + expected [][]int + }{ + { + name: "Simple", + input: []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 7, 2, 4, 5, 1, 4, 2}, + expected: [][]int{{1, 2, 3, 4}, {5}, {6}, {7}, {8}, {9}, {7, 2}, {4, 5, 1}, {4, 2}}, + }, + { + name: "Elements with larger than sum", + input: []int{1, 2, 3, 4, 5, 6, 11, 7, 8, 9, 7, 2, 50, 4, 5, 1, 4, 2, 23}, + expected: [][]int{{1, 2, 3, 4}, {5}, {6}, {11}, {7}, {8}, {9}, {7, 2}, {50}, {4, 5, 1}, {4, 2}, {23}}, + }, + } + + for _, test := range tests { + got := ChunkByMaxSum(test.input, 10) + assert.Equal(t, test.expected, got, test.name) + } +} + +func TestTimeToTimestamp(t *testing.T) { + tests := []struct { + name string + time string + expected int64 + }{ + { + name: "time string in CET location", + time: "2024-11-12T15:23:02+0100", + expected: 1731421382000, + }, + { + name: "time string in DST", + time: "2024-10-03T12:51:40+0200", + expected: 1727952700000, + }, + { + name: "time string in UTC", + time: "2024-11-12T15:23:02+0000", + expected: 1731424982000, + }, + } + + for _, test := range tests { + timeStamp := TimeToTimestamp(base.DatetimezoneLayout, test.time) + assert.Equal(t, test.expected, timeStamp, test.name) + } + + // Check failure case + timeStamp := TimeToTimestamp(base.DatetimezoneLayout, "Unknown") + assert.Equal(t, int64(0), timeStamp) +} + +func TestChunkBy(t *testing.T) { + tests := []struct { + name string + input []int + expected [][]int + size int + }{ + { + name: "chunk size less than length", + input: []int{1, 2, 3, 4, 5, 6}, + size: 3, + expected: [][]int{{1, 2, 3}, {4, 5, 6}}, + }, + { + name: "chunk size more than length", + input: []int{1, 2, 3, 4, 5, 6}, + size: 10, + expected: [][]int{{1, 2, 3, 4, 5, 6}}, + }, + { + name: "chunk size 0", + input: []int{1, 2, 3, 4, 5, 6}, + size: 0, + expected: [][]int{{1, 2, 3, 4, 5, 6}}, + }, + } + + for _, test := range tests { + got := ChunkBy(test.input, test.size) + assert.Equal(t, test.expected, got, test.name) + } +} diff --git a/pkg/api/helper/helper.go b/pkg/api/helper/helper.go deleted file mode 100644 index 6d11cba5..00000000 --- a/pkg/api/helper/helper.go +++ /dev/null @@ -1,30 +0,0 @@ -// Package helper provides utility functions across sub packages -package helper - -import ( - "time" -) - -// TimeToTimestamp converts a date in a given layout to unix timestamp of the date. -func TimeToTimestamp(layout string, date string) int64 { - t, err := time.Parse(layout, date) - if err == nil { - return t.UnixMilli() - } - - return 0 -} - -// ChunkBy splits the slice into chunks of given size. -func ChunkBy[T any](items []T, chunkSize int) [][]T { - if chunkSize == 0 { - return [][]T{items} - } - - _chunks := make([][]T, 0, (len(items)/chunkSize)+1) - for chunkSize < len(items) { - items, _chunks = items[chunkSize:], append(_chunks, items[0:chunkSize:chunkSize]) - } - - return append(_chunks, items) -} diff --git a/pkg/api/helper/helper_test.go b/pkg/api/helper/helper_test.go deleted file mode 100644 index 6f6e04fa..00000000 --- a/pkg/api/helper/helper_test.go +++ /dev/null @@ -1,74 +0,0 @@ -package helper - -import ( - "testing" - - "github.com/ceems-dev/ceems/pkg/api/base" - "github.com/stretchr/testify/assert" -) - -func TestTimeToTimestamp(t *testing.T) { - tests := []struct { - name string - time string - expected int64 - }{ - { - name: "time string in CET location", - time: "2024-11-12T15:23:02+0100", - expected: 1731421382000, - }, - { - name: "time string in DST", - time: "2024-10-03T12:51:40+0200", - expected: 1727952700000, - }, - { - name: "time string in UTC", - time: "2024-11-12T15:23:02+0000", - expected: 1731424982000, - }, - } - - for _, test := range tests { - timeStamp := TimeToTimestamp(base.DatetimezoneLayout, test.time) - assert.Equal(t, test.expected, timeStamp, test.name) - } - - // Check failure case - timeStamp := TimeToTimestamp(base.DatetimezoneLayout, "Unknown") - assert.Equal(t, int64(0), timeStamp) -} - -func TestChunkBy(t *testing.T) { - tests := []struct { - name string - input []int - expected [][]int - size int - }{ - { - name: "chunk size less than length", - input: []int{1, 2, 3, 4, 5, 6}, - size: 3, - expected: [][]int{{1, 2, 3}, {4, 5, 6}}, - }, - { - name: "chunk size more than length", - input: []int{1, 2, 3, 4, 5, 6}, - size: 10, - expected: [][]int{{1, 2, 3, 4, 5, 6}}, - }, - { - name: "chunk size 0", - input: []int{1, 2, 3, 4, 5, 6}, - size: 0, - expected: [][]int{{1, 2, 3, 4, 5, 6}}, - }, - } - - for _, test := range tests { - got := ChunkBy(test.input, test.size) - assert.Equal(t, test.expected, got, test.name) - } -} diff --git a/pkg/api/models/types.go b/pkg/api/models/types.go index 21841e64..1794a202 100644 --- a/pkg/api/models/types.go +++ b/pkg/api/models/types.go @@ -135,11 +135,18 @@ func (m MetricMap) Keys() []string { } // Values returns a slice of string representation of map values. -func (m MetricMap) Values(format string) []any { +func (m MetricMap) Values(format string, l int) []any { // Return empty string when map is nil // Useful in table generation for cacct app if len(m) == 0 { - return []any{""} + l = max(l, 1) + + s := make([]any, l) + for ik := range l { + s[ik] = "" + } + + return s } s := make([]any, len(m)) diff --git a/pkg/api/resource/default.go b/pkg/api/resource/default.go index c8023cb8..33886474 100644 --- a/pkg/api/resource/default.go +++ b/pkg/api/resource/default.go @@ -51,13 +51,5 @@ func (d *defaultResourceManager) FetchUsersProjects( ) ([]models.ClusterUsers, []models.ClusterProjects, error) { d.logger.Info("Empty users and projects fetched from default NoOp cluster") - return []models.ClusterUsers{ - { - Cluster: models.Cluster{ID: "default"}, - }, - }, []models.ClusterProjects{ - { - Cluster: models.Cluster{ID: "default"}, - }, - }, nil + return []models.ClusterUsers{{Cluster: models.Cluster{ID: "default"}}}, []models.ClusterProjects{{Cluster: models.Cluster{ID: "default"}}}, nil } diff --git a/pkg/api/resource/k8s/manager.go b/pkg/api/resource/k8s/manager.go index 2c554828..19a7ed52 100644 --- a/pkg/api/resource/k8s/manager.go +++ b/pkg/api/resource/k8s/manager.go @@ -171,11 +171,7 @@ func (k *k8sManager) FetchUsersProjects( // Fetch users and namespaces association userModels, projectModels := k.fetchUserNSs(ctx, currentTime) - return []models.ClusterUsers{ - {Cluster: k.cluster, Users: userModels}, - }, []models.ClusterProjects{ - {Cluster: k.cluster, Projects: projectModels}, - }, nil + return []models.ClusterUsers{{Cluster: k.cluster, Users: userModels}}, []models.ClusterProjects{{Cluster: k.cluster, Projects: projectModels}}, nil } func (k *k8sManager) fetchPods( diff --git a/pkg/api/resource/lsf/cli.go b/pkg/api/resource/lsf/cli.go index b94f21b9..667037bf 100644 --- a/pkg/api/resource/lsf/cli.go +++ b/pkg/api/resource/lsf/cli.go @@ -17,7 +17,6 @@ import ( "github.com/ceems-dev/ceems/internal/common" "github.com/ceems-dev/ceems/internal/osexec" "github.com/ceems-dev/ceems/pkg/api/base" - "github.com/ceems-dev/ceems/pkg/api/helper" "github.com/ceems-dev/ceems/pkg/api/models" ) @@ -154,7 +153,7 @@ func parseBacctCmdOutput(bacctOutput string, start time.Time, end time.Time, job components[c] = t.Format(base.DatetimezoneLayout) } - eventTS[c] = helper.TimeToTimestamp(base.DatetimezoneLayout, components[c]) + eventTS[c] = common.TimeToTimestamp(base.DatetimezoneLayout, components[c]) } // Get CPU allocations @@ -439,7 +438,7 @@ func parseBjobsCmdOutput(bjobsOutput []byte, start time.Time, end time.Time, job components[c] = "N/A" } - eventTS[c] = helper.TimeToTimestamp(base.DatetimezoneLayout, components[c]) + eventTS[c] = common.TimeToTimestamp(base.DatetimezoneLayout, components[c]) } // If job has already finished in the past bjobs should not return this job. diff --git a/pkg/api/resource/lsf/manager.go b/pkg/api/resource/lsf/manager.go index 166c3b27..d643581b 100644 --- a/pkg/api/resource/lsf/manager.go +++ b/pkg/api/resource/lsf/manager.go @@ -115,11 +115,7 @@ func (s *lsfScheduler) FetchUsersProjects( users, projects := s.fetchUserProjects(current) s.logger.Info("LSF user account data fetched", "cluster_id", s.cluster.ID, "num_users", len(users), "num_accounts", len(projects)) - return []models.ClusterUsers{ - {Cluster: s.cluster, Users: users, Append: true}, - }, []models.ClusterProjects{ - {Cluster: s.cluster, Projects: projects, Append: true}, - }, nil + return []models.ClusterUsers{{Cluster: s.cluster, Users: users, Append: true}}, []models.ClusterProjects{{Cluster: s.cluster, Projects: projects, Append: true}}, nil } return nil, nil, fmt.Errorf("unknown fetch mode for projects for LSF cluster %s", s.cluster.ID) diff --git a/pkg/api/resource/manager.go b/pkg/api/resource/manager.go index 02bcece5..72f92188 100644 --- a/pkg/api/resource/manager.go +++ b/pkg/api/resource/manager.go @@ -201,7 +201,7 @@ func (b Manager) FetchUnits(ctx context.Context, start time.Time, end time.Time) var wg sync.WaitGroup - wg.Add((len(b.Fetchers))) + wg.Add(len(b.Fetchers)) for _, fetcher := range b.Fetchers { go func(f Fetcher) { @@ -247,7 +247,7 @@ func (b Manager) FetchUsersProjects( var wg sync.WaitGroup - wg.Add((len(b.Fetchers))) + wg.Add(len(b.Fetchers)) for _, fetcher := range b.Fetchers { go func(f Fetcher) { diff --git a/pkg/api/resource/manager_test.go b/pkg/api/resource/manager_test.go index d2474deb..4453c46c 100644 --- a/pkg/api/resource/manager_test.go +++ b/pkg/api/resource/manager_test.go @@ -54,25 +54,7 @@ func (d *mockResourceManager) FetchUsersProjects( _ context.Context, currentTime time.Time, ) ([]models.ClusterUsers, []models.ClusterProjects, error) { - return []models.ClusterUsers{ - { - Cluster: models.Cluster{ID: "mock"}, - Users: []models.User{ - { - Name: "foo", - }, - }, - }, - }, []models.ClusterProjects{ - { - Cluster: models.Cluster{ID: "mock"}, - Projects: []models.Project{ - { - Name: "fooprj", - }, - }, - }, - }, nil + return []models.ClusterUsers{{Cluster: models.Cluster{ID: "mock"}, Users: []models.User{{Name: "foo"}}}}, []models.ClusterProjects{{Cluster: models.Cluster{ID: "mock"}, Projects: []models.Project{{Name: "fooprj"}}}}, nil } func mockConfig(tmpDir string, cfg string) string { diff --git a/pkg/api/resource/openstack/identity.go b/pkg/api/resource/openstack/identity.go index 867c2791..d490dd2f 100644 --- a/pkg/api/resource/openstack/identity.go +++ b/pkg/api/resource/openstack/identity.go @@ -10,8 +10,8 @@ import ( "sync" "time" + "github.com/ceems-dev/ceems/internal/common" "github.com/ceems-dev/ceems/pkg/api/base" - "github.com/ceems-dev/ceems/pkg/api/helper" "github.com/ceems-dev/ceems/pkg/api/models" ) @@ -144,7 +144,7 @@ func (o *openstackManager) usersProjectsAssoc(ctx context.Context, current time. // Chunk by userIDs in chunks of of a given size so that we make // concurrent corresponding to chunkSize each time to get projects // of each user - userIDChunks := helper.ChunkBy(userIDs, chunkSize) + userIDChunks := common.ChunkBy(userIDs, chunkSize) // Get user projects userProjects := make(map[string][]Project, len(userIDs)) diff --git a/pkg/api/resource/openstack/manager.go b/pkg/api/resource/openstack/manager.go index ac411fce..5197e01d 100644 --- a/pkg/api/resource/openstack/manager.go +++ b/pkg/api/resource/openstack/manager.go @@ -222,11 +222,7 @@ func (o *openstackManager) FetchUsersProjects( } } - return []models.ClusterUsers{ - {Cluster: o.cluster, Users: o.userProjectsCache.userModels}, - }, []models.ClusterProjects{ - {Cluster: o.cluster, Projects: o.userProjectsCache.projectModels}, - }, nil + return []models.ClusterUsers{{Cluster: o.cluster, Users: o.userProjectsCache.userModels}}, []models.ClusterProjects{{Cluster: o.cluster, Projects: o.userProjectsCache.projectModels}}, nil } // servers endpoint. diff --git a/pkg/api/resource/openstack/request.go b/pkg/api/resource/openstack/request.go index c9512b07..20e933fa 100644 --- a/pkg/api/resource/openstack/request.go +++ b/pkg/api/resource/openstack/request.go @@ -14,7 +14,7 @@ func apiRequest[T any](req *http.Request, client *http.Client) (T, error) { req.Header.Add("Content-Type", "application/json") // Make request - resp, err := client.Do(req) //nolint:gosec + resp, err := client.Do(req) if err != nil { return *new(T), err } @@ -48,7 +48,7 @@ func apiTokenRequest(req *http.Request, client *http.Client) (string, error) { req.Header.Add("Content-Type", "application/json") // Make request - resp, err := client.Do(req) //nolint:gosec + resp, err := client.Do(req) if err != nil { return "", err } diff --git a/pkg/api/resource/slurm/cli.go b/pkg/api/resource/slurm/cli.go index 6965a8ac..d5213c96 100644 --- a/pkg/api/resource/slurm/cli.go +++ b/pkg/api/resource/slurm/cli.go @@ -18,7 +18,6 @@ import ( internal_osexec "github.com/ceems-dev/ceems/internal/osexec" "github.com/ceems-dev/ceems/internal/security" "github.com/ceems-dev/ceems/pkg/api/base" - "github.com/ceems-dev/ceems/pkg/api/helper" "github.com/ceems-dev/ceems/pkg/api/models" "kernel.org/pub/linux/libs/security/libcap/cap" ) @@ -212,7 +211,7 @@ func parseSacctCmdOutput(sacctOutput string, start time.Time, end time.Time) ([] } } - eventTS[c] = helper.TimeToTimestamp(base.DatetimezoneLayout, components[sacctFieldMap[c]]) + eventTS[c] = common.TimeToTimestamp(base.DatetimezoneLayout, components[sacctFieldMap[c]]) } // Parse alloctres to get billing, nnodes, ncpus, ngpus and mem diff --git a/pkg/api/resource/slurm/manager.go b/pkg/api/resource/slurm/manager.go index f0d7c89f..4421f39f 100644 --- a/pkg/api/resource/slurm/manager.go +++ b/pkg/api/resource/slurm/manager.go @@ -129,11 +129,7 @@ func (s *slurmScheduler) FetchUsersProjects( return nil, nil, err } - return []models.ClusterUsers{ - {Cluster: s.cluster, Users: users}, - }, []models.ClusterProjects{ - {Cluster: s.cluster, Projects: projects}, - }, nil + return []models.ClusterUsers{{Cluster: s.cluster, Users: users}}, []models.ClusterProjects{{Cluster: s.cluster, Projects: projects}}, nil } return nil, nil, fmt.Errorf("unknown fetch mode for projects for SLURM cluster %s", s.cluster.ID) diff --git a/pkg/api/updater/tsdb/tsdb.go b/pkg/api/updater/tsdb/tsdb.go index 1073d9ca..f74dd749 100644 --- a/pkg/api/updater/tsdb/tsdb.go +++ b/pkg/api/updater/tsdb/tsdb.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "log/slog" - "maps" "math" "strings" "sync" @@ -15,7 +14,6 @@ import ( "github.com/ceems-dev/ceems/internal/common" "github.com/ceems-dev/ceems/pkg/api/base" - "github.com/ceems-dev/ceems/pkg/api/helper" "github.com/ceems-dev/ceems/pkg/api/models" "github.com/ceems-dev/ceems/pkg/api/updater" "github.com/ceems-dev/ceems/pkg/tsdb" @@ -280,8 +278,8 @@ func (t *tsdbUpdater) fetchAggMetrics( duration time.Duration, uuids []string, settings *tsdb.Settings, -) map[string]map[string]tsdb.Metric { - aggMetrics := make(map[string]map[string]tsdb.Metric, len(t.config.Queries)) +) map[string]map[string]model.Vector { + aggMetrics := make(map[string]map[string]model.Vector, len(t.config.Queries)) // If duration is less than rateInterval bail if duration < settings.RateInterval { @@ -325,7 +323,7 @@ func (t *tsdbUpdater) fetchAggMetrics( go func(n string, sn string, q string) { defer wg.Done() - var aggMetric tsdb.Metric + var aggMetric model.Vector var err error @@ -350,7 +348,7 @@ func (t *tsdbUpdater) fetchAggMetrics( metricLock.Lock() if aggMetrics[n] == nil { - aggMetrics[n] = make(map[string]tsdb.Metric) + aggMetrics[n] = make(map[string]model.Vector) } aggMetrics[n][sn] = aggMetric @@ -456,10 +454,10 @@ func (t *tsdbUpdater) update( // Batch UUIDs into slices of 1000 so that we make TSDB requests for each 1000 units // This is to safeguard against OOM errors due to a very large number of units // that can spread across big time interval - uuidBatches := helper.ChunkBy(allUnitUUIDs[:j], batchSize) + uuidBatches := common.ChunkBy(allUnitUUIDs[:j], batchSize) numBatches := len(uuidBatches) - aggMetrics := make(map[string]map[string]tsdb.Metric) + aggMetrics := make(map[string]map[string]map[string]float64) // Loop over each chunk for iBatch, batchUUIDs := range uuidBatches { @@ -478,21 +476,43 @@ func (t *tsdbUpdater) update( // If inner map has not been initialized yet, do it // These are parent metrics like avg_cpu_usage, avg_gpu_usage if aggMetrics[metricName] == nil { - aggMetrics[metricName] = make(map[string]tsdb.Metric, len(metrics)) + aggMetrics[metricName] = make(map[string]map[string]float64, len(metrics)) } // Each parent metric has sub metrics that operator chooses and we loop // over them here for subMetricName, subMetrics := range metrics { if aggMetrics[metricName][subMetricName] == nil { - aggMetrics[metricName][subMetricName] = make(tsdb.Metric, len(subMetrics)) + aggMetrics[metricName][subMetricName] = make(map[string]float64, len(subMetrics)) } - maps.Copy(aggMetrics[metricName][subMetricName], subMetrics) + // Check how many labels each metric returned. We expect it to return + // exactly one label per metric which is uuid. If it returns more than + // 1, it means there is a query that is aggregated by more than 1 label + // which is not supported. If that is the case, emit a warning log + var subMetricLabels []string + + // Extract uuid label from model.Vector + for _, value := range subMetrics { + if len(value.Metric) > len(subMetricLabels) { + for l := range value.Metric { + subMetricLabels = append(subMetricLabels, string(l)) + } + } + + if uuid, ok := value.Metric["uuid"]; ok { + aggMetrics[metricName][subMetricName][string(uuid)] = float64(value.Value) + } + } + + // If number of labels is more than 1 emit a warning log + if len(subMetricLabels) > 1 { + t.Logger.Warn("TSDB query must be aggregated over uuid and return only uuid label in result. More than 1 labels found in query results.", "metric", metricName, "sub_metric", subMetricName, "found_labels", strings.Join(subMetricLabels, ",")) + } } } t.Logger.Debug( - "progress", "batch_id", iBatch, "total_batches", numBatches, "batch_size", batchSize, + "Progress", "batch_id", iBatch, "total_batches", numBatches, "batch_size", batchSize, ) } } @@ -685,7 +705,7 @@ func (t *tsdbUpdater) deleteTimeSeries( // // Join them with | as delimiter. We will use regex match to match all series // with the label uuid=~"$unitids" - allUUIDs := strings.Join(unitUUIDs, "|") + allUUIDs := strings.ReplaceAll(strings.ReplaceAll(strings.Join(unitUUIDs, "|"), "[", `\[`), "]", `\]`) matchers := t.config.LabelsToDrop matchers = append(matchers, fmt.Sprintf("{uuid=~\"%s\"}", allUUIDs)) diff --git a/pkg/collector/cgroup.go b/pkg/collector/cgroup.go index de38f2b6..7a147697 100644 --- a/pkg/collector/cgroup.go +++ b/pkg/collector/cgroup.go @@ -1323,7 +1323,7 @@ func (c *cgroupCollector) cpusFromChildren(path string) (int, error) { // In cgroup v1, they are flat whereas in cgroup v2 they are inside libvirt folder var vcpuPath string - if c.cgroupManager.mode == cgroups.Unified && !(c.cgroupManager.nonSystemdLayout) { + if c.cgroupManager.mode == cgroups.Unified && !c.cgroupManager.nonSystemdLayout { vcpuPath = fmt.Sprintf("%s%s/libvirt/vcpu*", c.cgroupManager.root, path) } else { vcpuPath = fmt.Sprintf("%s%s/vcpu*", c.cgroupManager.root, path) diff --git a/pkg/collector/ebpf.go b/pkg/collector/ebpf.go index a03c9a30..5d5ce5b8 100644 --- a/pkg/collector/ebpf.go +++ b/pkg/collector/ebpf.go @@ -1123,8 +1123,7 @@ func loadObject(path string) (*ebpf.Collection, error) { spec, err := ebpf.LoadCollectionSpecFromReader(reader) if err != nil { - var ve *ebpf.VerifierError - if errors.As(err, &ve) { + if ve, ok := errors.AsType[*ebpf.VerifierError](err); ok { err = fmt.Errorf("%+v", ve) //nolint:errorlint } @@ -1134,8 +1133,7 @@ func loadObject(path string) (*ebpf.Collection, error) { // Instantiate a Collection from a CollectionSpec. coll, err := ebpf.NewCollection(spec) if err != nil { - var ve *ebpf.VerifierError - if errors.As(err, &ve) { + if ve, ok := errors.AsType[*ebpf.VerifierError](err); ok { err = fmt.Errorf("%+v", ve) //nolint:errorlint } diff --git a/pkg/collector/hwmon.go b/pkg/collector/hwmon.go index 1e22be70..9e8ff131 100644 --- a/pkg/collector/hwmon.go +++ b/pkg/collector/hwmon.go @@ -422,7 +422,7 @@ func sysReadFile(file string) ([]byte, error) { // From docs: // The int, uint, and uintptr types are usually 32 bits wide on 32-bit systems and 64 bits wide on 64-bit systems. // Safe to ignore this gosec warning - n, err := unix.Read(int(f.Fd()), b) //nolint:gosec + n, err := unix.Read(int(f.Fd()), b) if err != nil { return nil, err } diff --git a/pkg/lb/frontend/helpers.go b/pkg/lb/frontend/helpers.go index f7faa66e..e6b63a7f 100644 --- a/pkg/lb/frontend/helpers.go +++ b/pkg/lb/frontend/helpers.go @@ -13,7 +13,7 @@ func ceemsAPIRequest[T any](req *http.Request, client *http.Client) ([]T, error) // Make request // If request failed, forbid the query. It can happen when CEEMS API server // goes offline and we should wait for it to come back online - resp, err := client.Do(req) //nolint:gosec + resp, err := client.Do(req) if err != nil { return nil, err } else { diff --git a/pkg/tsdb/client.go b/pkg/tsdb/client.go index 5d51e8bc..96f067fd 100644 --- a/pkg/tsdb/client.go +++ b/pkg/tsdb/client.go @@ -30,7 +30,7 @@ var ( var settingsLock = sync.RWMutex{} // Metric defines Client metrics. -type Metric map[string]float64 +// type Metric map[string]float64 // RangeMetric defines Client range metrics. // type RangeMetric map[string][]model.SamplePair @@ -224,7 +224,7 @@ func (t *Client) Labels(ctx context.Context, matchers []string, start time.Time, } // Query makes a Client query. -func (t *Client) Query(ctx context.Context, query string, queryTime time.Time, timeout time.Duration) (Metric, error) { +func (t *Client) Query(ctx context.Context, query string, queryTime time.Time, timeout time.Duration) (model.Vector, error) { // Get current scrape interval to use as lookback_delta // This query parameter is undocumented on Prometheus. If we use // default value of 5m, we tend to have metrics 5m **after** compute @@ -250,7 +250,7 @@ func (t *Client) Query(ctx context.Context, query string, queryTime time.Time, t } // Parse data - queriedValues := make(Metric) + // queriedValues := make(Metric) var values model.Vector @@ -261,14 +261,14 @@ func (t *Client) Query(ctx context.Context, query string, queryTime time.Time, t return nil, fmt.Errorf("%w on data: %v", ErrFailedTypeAssertion, result) } - // Iterate over each value and make UUID to value map - for _, value := range values { - if uuid, ok := value.Metric["uuid"]; ok { - queriedValues[string(uuid)] = float64(value.Value) - } - } + // // Iterate over each value and make UUID to value map + // for _, value := range values { + // if uuid, ok := value.Metric["uuid"]; ok { + // queriedValues[string(uuid)] = float64(value.Value) + // } + // } - return queriedValues, nil + return values, nil } // RangeQuery makes a Client range query. diff --git a/pkg/tsdb/client_test.go b/pkg/tsdb/client_test.go index 31bc201b..d4592a75 100644 --- a/pkg/tsdb/client_test.go +++ b/pkg/tsdb/client_test.go @@ -357,7 +357,14 @@ func TestTSDBQuerySuccess(t *testing.T) { m, err := tsdb.Query(t.Context(), "foo", time.Now(), defaultQueryTimeout) require.NoError(t, err) - assert.Equal(t, Metric{"1": 1.1, "2": 2.2}, m) + + // Extract uuid from vector + mmap := make(map[string]float64) + for _, v := range m { + mmap[string(v.Metric["uuid"])] = float64(v.Value) + } + + assert.Equal(t, map[string]float64{"1": 1.1, "2": 2.2}, mmap) assert.Equal(t, 15*time.Second, time.Duration(expectedQueryLookback)) } diff --git a/scripts/e2e-test.sh b/scripts/e2e-test.sh index eed09d23..956be62f 100755 --- a/scripts/e2e-test.sh +++ b/scripts/e2e-test.sh @@ -421,6 +421,10 @@ then then desc="cacct with default format" fixture='cmd/cacct/testdata/output/e2e-test-cacct-default-format.txt' + elif [ "${scenario}" = "cacct-help-format" ] + then + desc="cacct with help format" + fixture='cmd/cacct/testdata/output/e2e-test-cacct-help-format.txt' elif [ "${scenario}" = "cacct-long-format" ] then desc="cacct with long format" @@ -445,6 +449,10 @@ then then desc="cacct using invalid config" fixture='cmd/cacct/testdata/output/e2e-test-cacct-invalid-config.txt' + elif [ "${scenario}" = "cacct-bad-config" ] + then + desc="cacct using bad config" + fixture='cmd/cacct/testdata/output/e2e-test-cacct-bad-config.txt' elif [ "${scenario}" = "cacct-tsdata" ] then desc="cacct to dump time series data" @@ -1646,12 +1654,15 @@ then if [ "${scenario}" = "cacct-default-format" ] then ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2022-03-20" > "${fixture_output}" 2>&1 + elif [ "${scenario}" = "cacct-help-format" ] + then + ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --helpformat > "${fixture_output}" 2>&1 elif [ "${scenario}" = "cacct-long-format" ] then ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2022-03-20" --long > "${fixture_output}" 2>&1 elif [ "${scenario}" = "cacct-custom-format" ] then - ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2022-03-20" --format="jobid,account,cpuusage" > "${fixture_output}" 2>&1 + ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2022-03-20" --format="jobid,account,cpuusage,avgUsageSomethingScalar,avgUsageSomethingVector" > "${fixture_output}" 2>&1 elif [ "${scenario}" = "cacct-admin-user" ] then ./bin/cacct --current-user=grafana --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2022-03-20" --user=usr1,usr2 > "${fixture_output}" 2>&1 @@ -1664,16 +1675,19 @@ then elif [ "${scenario}" = "cacct-invalid-config" ] then ./bin/cacct --current-user=usr1 --config-path="nonexistant/testdata" --starttime="2022-02-20" --endtime="2022-03-20" > "${fixture_output}" 2>&1 || true + elif [ "${scenario}" = "cacct-bad-config" ] + then + ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata/bad-config" --starttime="2022-02-20" --endtime="2022-03-20" > "${fixture_output}" 2>&1 || true elif [ "${scenario}" = "cacct-tsdata" ] then - ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --job="147973" --ts --ts.out-dir="${tmpdir}/ts" > "${fixture_output}" 2>&1 + ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --job="147973" --ts.metrics="cpuusage" --ts.out-dir="${tmpdir}/ts" > "${fixture_output}" 2>&1 cat "${tmpdir}/ts/metadata.json" >> "${fixture_output}" cat "${tmpdir}/ts/554b56cadf9dea4b.csv" >> "${fixture_output}" # Remove line that says time series data has been saved in the output file as it will contain directory name which changes on every run sed -i '/^time series data saved to directory/d' "${fixture_output}" elif [ "${scenario}" = "cacct-tsdata-fail" ] then - ./bin/cacct --current-user=usr1 --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2022-03-20" --ts --ts.out-dir="${tmpdir}/ts" > "${fixture_output}" 2>&1 || true + ./bin/cacct --current-user=grafana --user=all --config-path="cmd/cacct/testdata" --starttime="2022-02-20" --endtime="2024-03-20" --ts.metrics="cpuusage" --ts.out-dir="${tmpdir}/ts" > "${fixture_output}" 2>&1 || true fi elif [[ "${scenario}" =~ ^"tool" ]] diff --git a/scripts/mock_exporters/main.go b/scripts/mock_exporters/main.go index 94f9bc5d..11278b4f 100644 --- a/scripts/mock_exporters/main.go +++ b/scripts/mock_exporters/main.go @@ -48,7 +48,7 @@ type dcgmCollector struct { } func randFloat(minVal, maxVal float64) float64 { - return minVal + rand.Float64()*(maxVal-minVal) //nolint:gosec + return minVal + rand.Float64()*(maxVal-minVal) } func newDCGMCollector() *dcgmCollector { @@ -185,17 +185,17 @@ func (collector *dcgmCollector) Collect(ch chan<- prometheus.Metric) { for _, dev := range collector.devices { ch <- prometheus.MustNewConstMetric( - collector.gpuUtil, prometheus.GaugeValue, 100*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuUtil, prometheus.GaugeValue, 100*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuMemUsed, prometheus.GaugeValue, 100*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuMemUsed, prometheus.GaugeValue, 100*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuMemFree, prometheus.GaugeValue, 100*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuMemFree, prometheus.GaugeValue, 100*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) @@ -210,62 +210,62 @@ func (collector *dcgmCollector) Collect(ch chan<- prometheus.Metric) { ) ch <- prometheus.MustNewConstMetric( - collector.gpuSMActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuSMActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuSMOcc, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuSMOcc, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuGREngActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuGREngActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuPipeActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuPipeActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuFP64Active, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuFP64Active, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuFP32Active, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuFP32Active, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuFP16Active, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuFP16Active, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuDRAMActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuDRAMActive, prometheus.GaugeValue, rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuNVLRX, prometheus.GaugeValue, 1024*1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuNVLRX, prometheus.GaugeValue, 1024*1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuNVLTX, prometheus.GaugeValue, 1024*1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuNVLTX, prometheus.GaugeValue, 1024*1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuPCIeTX, prometheus.GaugeValue, 1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuPCIeTX, prometheus.GaugeValue, 1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) ch <- prometheus.MustNewConstMetric( - collector.gpuPCIeRX, prometheus.GaugeValue, 1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, //nolint:gosec + collector.gpuPCIeRX, prometheus.GaugeValue, 1024*1024*rand.Float64(), "host", dev.UUID, dev.IID, "nvidia"+dev.ID, dev.ID, dev.PCIAddr, "NVIDIA A100 80GiB", ) } @@ -317,12 +317,12 @@ func (collector *amdSMICollector) Describe(ch chan<- *prometheus.Desc) { func (collector *amdSMICollector) Collect(ch chan<- prometheus.Metric) { for idev := range collector.devices { ch <- prometheus.MustNewConstMetric( - collector.gpuUtil, prometheus.GaugeValue, 100*rand.Float64(), strconv.Itoa(idev), //nolint:gosec + collector.gpuUtil, prometheus.GaugeValue, 100*rand.Float64(), strconv.Itoa(idev), "Advanced Micro Devices Inc", ) ch <- prometheus.MustNewConstMetric( - collector.gpuMemUtil, prometheus.GaugeValue, 100*rand.Float64(), strconv.Itoa(idev), //nolint:gosec + collector.gpuMemUtil, prometheus.GaugeValue, 100*rand.Float64(), strconv.Itoa(idev), "Advanced Micro Devices Inc", ) // GPU power reported in micro Watts @@ -478,7 +478,7 @@ func (collector *amdDeviceMetricsCollector) Describe(ch chan<- *prometheus.Desc) func (collector *amdDeviceMetricsCollector) Collect(ch chan<- prometheus.Metric) { for _, dev := range collector.devices { ch <- prometheus.MustNewConstMetric( - collector.gpuUtil, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuUtil, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) @@ -488,7 +488,7 @@ func (collector *amdDeviceMetricsCollector) Collect(ch chan<- prometheus.Metric) ) ch <- prometheus.MustNewConstMetric( - collector.gpuVRAMUsed, prometheus.GaugeValue, 1024*1024*1024*24*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuVRAMUsed, prometheus.GaugeValue, 1024*1024*1024*24*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) @@ -498,7 +498,7 @@ func (collector *amdDeviceMetricsCollector) Collect(ch chan<- prometheus.Metric) ) ch <- prometheus.MustNewConstMetric( - collector.gpuGTTUsed, prometheus.GaugeValue, 1024*1024*24*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuGTTUsed, prometheus.GaugeValue, 1024*1024*24*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) @@ -508,7 +508,7 @@ func (collector *amdDeviceMetricsCollector) Collect(ch chan<- prometheus.Metric) ) ch <- prometheus.MustNewConstMetric( - collector.gpuVisibleRAMUsed, prometheus.GaugeValue, 1024*1024*24*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuVisibleRAMUsed, prometheus.GaugeValue, 1024*1024*24*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) @@ -523,42 +523,42 @@ func (collector *amdDeviceMetricsCollector) Collect(ch chan<- prometheus.Metric) ) ch <- prometheus.MustNewConstMetric( - collector.gpuSMActive, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuSMActive, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuTensorActive, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuTensorActive, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuProfOccupancy, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuProfOccupancy, prometheus.GaugeValue, 100*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuFP64Ops, prometheus.GaugeValue, 1e8*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuFP64Ops, prometheus.GaugeValue, 1e8*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuFP32Ops, prometheus.GaugeValue, 1e6*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuFP32Ops, prometheus.GaugeValue, 1e6*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuFP16Ops, prometheus.GaugeValue, 1e3*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuFP16Ops, prometheus.GaugeValue, 1e3*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuWriteSize, prometheus.GaugeValue, 5e6*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuWriteSize, prometheus.GaugeValue, 5e6*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) ch <- prometheus.MustNewConstMetric( - collector.gpuReadSize, prometheus.GaugeValue, 3e6*rand.Float64(), dev.ID, //nolint:gosec + collector.gpuReadSize, prometheus.GaugeValue, 3e6*rand.Float64(), dev.ID, dev.IID, dev.UUID, ) } diff --git a/scripts/mock_servers/main.go b/scripts/mock_servers/main.go index f638a0fd..0cb484da 100644 --- a/scripts/mock_servers/main.go +++ b/scripts/mock_servers/main.go @@ -167,6 +167,17 @@ func ProfilesHandler(w http.ResponseWriter, r *http.Request) { w.Write([]byte("KO")) } +// containsSubString will check if any of elements in l contains s. +func containsSubstring(l []string, s string) bool { + for _, e := range l { + if strings.Contains(s, e) { + return true + } + } + + return false +} + // QueryHandler handles queries. func QueryHandler(w http.ResponseWriter, r *http.Request) { var response tsdb.Response[any] @@ -214,12 +225,12 @@ func QueryHandler(w http.ResponseWriter, r *http.Request) { query = matches[1] } - // log.Println("Query", query, "UUIDs", uuids) + // log.Println("Query", query, "UUIDs", uuids, query) var results []any switch { - case slices.Contains( + case containsSubstring( []string{ "avg_cpu_usage", "avg_cpu_mem_usage", "avg_gpu_usage", "avg_gpu_mem_usage", "total_cpu_energy_usage_kwh", @@ -242,7 +253,7 @@ func QueryHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_io_read_stats_bytes", "total_io_write_stats_bytes", }, query): @@ -259,7 +270,7 @@ func QueryHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_io_read_stats_requests", "total_io_write_stats_requests", }, query): @@ -276,7 +287,7 @@ func QueryHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_ingress_stats_bytes", "total_egress_stats_bytes", }, query): @@ -293,7 +304,7 @@ func QueryHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_ingress_stats_packets", "total_egress_stats_packets", }, query): @@ -379,7 +390,7 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) { status := "success" switch { - case slices.Contains( + case containsSubstring( []string{ "avg_cpu_usage", "avg_cpu_mem_usage", "avg_gpu_usage", "avg_gpu_mem_usage", "total_cpu_energy_usage_kwh", "total_gpu_energy_usage_kwh", @@ -404,7 +415,7 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_io_read_stats_bytes", "total_io_write_stats_bytes", }, query): @@ -424,7 +435,7 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_io_read_stats_requests", "total_io_write_stats_requests", }, query): @@ -444,7 +455,7 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_ingress_stats_bytes", "total_egress_stats_bytes", }, query): @@ -464,7 +475,7 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) { }, }) } - case slices.Contains( + case containsSubstring( []string{ "total_ingress_stats_packets", "total_egress_stats_packets", }, query): diff --git a/scripts/pyro_requestor/main.go b/scripts/pyro_requestor/main.go index 6f0f0547..740702fe 100644 --- a/scripts/pyro_requestor/main.go +++ b/scripts/pyro_requestor/main.go @@ -54,7 +54,7 @@ func main() { log.Fatalln("failed to marshal message", err) } - req, err := http.NewRequest(http.MethodPost, *url, bytes.NewBuffer(data)) //nolint:noctx + req, err := http.NewRequest(http.MethodPost, *url, bytes.NewBuffer(data)) if err != nil { log.Fatalln("failed to create new request", err) } diff --git a/website/docs/cli/cacct.md b/website/docs/cli/cacct.md index addbdaf7..ae026632 100644 --- a/website/docs/cli/cacct.md +++ b/website/docs/cli/cacct.md @@ -17,7 +17,7 @@ sidebar_position: 4 | `--user` | Comma separated list of user names to select jobs to display.A special value `all` can be used to fetch jobs of all users when querying user has enough privileges. By default, the running user is used. | | | `--format` | Comma separated list of fields | | | `--helpformat` | List of available fields | | -| `--ts` | Time series data of jobs are saved in CSV format | `false` | +| `--ts.metrics` | Comma separated list of time series metrics. Check available metrics using `--helpformat` flag. | | | `--ts.out-dir` | Directory to save time series data | `out` | | `--csv` | Produce CSV output | `false` | | `--html` | Produce HTML output | `false` | diff --git a/website/docs/configuration/cacct.md b/website/docs/configuration/cacct.md index 7b1e9ae3..7d862de8 100644 --- a/website/docs/configuration/cacct.md +++ b/website/docs/configuration/cacct.md @@ -86,46 +86,104 @@ tsdb: basic_auth: username: prometheus password: anothersupersecretpassword + max_units_for_range_queries: 5 + scrape_interval: 10s + evaluation_interval: 10s queries: # CPU utilization - cpu_usage: uuid:ceems_cpu_usage:ratio_irate{uuid=~"%s"} + - name: cpu_usage + title: "CPU Usage (%)" + help: "Usage of CPUs in %" + query: uuid:ceems_cpu_usage:ratio_irate{uuid=~`{{.UUIDs}}`} + kind: range # CPU Memory utilization - cpu_mem_usage: uuid:ceems_cpu_memory_usage:ratio{uuid=~"%s"} + - name: cpu_mem_usage + title: "CPU Memory Usage (%)" + help: "Ratio of memory used to memory reserved of CPU in %" + query: uuid:ceems_cpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} + kind: range # Host power usage in Watts - host_power_usage: uuid:ceems_host_power_watts:pue{uuid=~"%s"} + - name: host_power_usage + title: "Host Power usage (W)" + help: "Instanteous power usage of host in Watts" + query: uuid:ceems_host_power_watts:pue{uuid=~`{{.UUIDs}}`} + kind: range # Host emissions in g/s - host_emissions: uuid:ceems_host_emissions_g_s:pue{uuid=~"%s"} + - name: host_emissions + title: "Host Eq. Emissions Rate (g/s)" + help: "Instanteous emissions rate of host in g/s" + query: uuid:ceems_host_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} + kind: range # GPU utilization - avg_gpu_usage: uuid:ceems_gpu_usage:ratio{uuid=~"%s"} + - name: avg_gpu_usage + title: "GPU Usage (%)" + help: "Usage of GPU in %" + query: uuid:ceems_gpu_usage:ratio{uuid=~`{{.UUIDs}}`} + kind: range # GPU memory utilization - avg_gpu_mem_usage: uuid:ceems_gpu_memory_usage:ratio{uuid=~"%s"} + - name: avg_gpu_mem_usage + title: "GPU Memory Usage (%)" + help: "Ratio of memory used to memory reserved of GPU in %" + query: uuid:ceems_gpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} + kind: range # GPU power usage in Watts - gpu_power_usage: uuid:ceems_gpu_power_watts:pue{uuid=~"%s"} + - name: gpu_power_usage + title: "GPU Power Usage (W)" + help: "Instanteous Power Usage of GPU in Watts" + query: uuid:ceems_gpu_power_watts:pue{uuid=~`{{.UUIDs}}`} + kind: range # GPU emissions in g/s - gpu_emissions: uuid:ceems_gpu_emissions_g_s:pue{uuid=~"%s"} + - name: gpu_emissions + title: "GPU Eq. Emission Rate (g/s)" + help: "Instanteous emissions rate of GPU in g/s" + query: uuid:ceems_gpu_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} + kind: range # Read IO bytes/s - io_read_bytes: irate(ceems_ebpf_read_bytes_total{uuid=~"%s"}[1m]) + - name: io_read_bytes + title: "IO Read Bandwidth (b/s)" + help: "Instanteous IO read bandwidth in bytes/s" + query: irate(ceems_ebpf_read_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) + kind: range # Write IO bytes/s - io_write_bytes: irate(ceems_ebpf_write_bytes_total{uuid=~"%s"}[1m]) + - name: io_write_bytes + title: "IO Write Bandwidth (b/s)" + help: "Instanteous IO write bandwidth in bytes/s" + query: irate(ceems_ebpf_write_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) + kind: range ``` +The keys in the `tsdb` section are explained as below: + +- `tsdb.max_units_for_range_queries`: Maximum number of units' time series to be fetched in a single +CLI execution. Use an appropriate number based on the deployment as making too many +range queries to TSDB can spike the memory usage of TSDB server. Default value is 10. +- `tsdb.scrape_interval`: The scrape interval of the TSDB jobs where queries are being +executed +- `tsdb.evaluation_interval`: The evaluation interval of the TSDB recording rules +- `tsdb.queries`: A list of queries to be executed to fetch the time series data. + - `tsdb.queries.name`: An **unique** short name for the query + - `tsdb.queries.title`: A human readable short title for the query. It will be included in the + output, so choose a name easy to understand for the end users. + - `tsdb.queries.help`: A small help text to explain what query metrics means + - `tsdb.queries.query`: It is the TSDB query that will be executed. Available + template variales are `{{.UUIDs}}`, `{{.ScrapeInterval}}`, `{{.RateInterval}}` and `{{.Range}}`. + - `tsdb.queries.kind`: Type of TSDB query. Currently only `range` is supported. + Similar to the CEEMS API server configuration, this example assumes the TSDB server is reachable at `tsdb:9090` and basic authentication is configured on the HTTP server. The `tsdb.queries` section is where operators configure the queries to pull time series data for each metric. If operators used [`ceems_tool`](../usage/ceems-tool.md) to generate recording rules for the TSDB, the queries in the sample configuration above will work -out-of-the-box. The keys in the `queries` object can be chosen freely; they are provided -for configuration file maintainability. The placeholder `%s` will be replaced by the compute -unit UUIDs at runtime before executing the queries on the TSDB server. +out-of-the-box. :::note[NOTE] @@ -167,36 +225,79 @@ tsdb: basic_auth: username: prometheus password: anothersupersecretpassword + max_units_for_range_queries: 5 + scrape_interval: 10s + evaluation_interval: 10s queries: # CPU utilization - cpu_usage: uuid:ceems_cpu_usage:ratio_irate{uuid=~"%s"} + - name: cpu_usage + title: "CPU Usage (%)" + help: "Usage of CPUs in %" + query: uuid:ceems_cpu_usage:ratio_irate{uuid=~`{{.UUIDs}}`} + kind: range # CPU Memory utilization - cpu_mem_usage: uuid:ceems_cpu_memory_usage:ratio{uuid=~"%s"} + - name: cpu_mem_usage + title: "CPU Memory Usage (%)" + help: "Ratio of memory used to memory reserved of CPU in %" + query: uuid:ceems_cpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} + kind: range # Host power usage in Watts - host_power_usage: uuid:ceems_host_power_watts:pue{uuid=~"%s"} + - name: host_power_usage + title: "Host Power usage (W)" + help: "Instanteous power usage of host in Watts" + query: uuid:ceems_host_power_watts:pue{uuid=~`{{.UUIDs}}`} + kind: range # Host emissions in g/s - host_emissions: uuid:ceems_host_emissions_g_s:pue{uuid=~"%s"} + - name: host_emissions + title: "Host Eq. Emissions Rate (g/s)" + help: "Instanteous emissions rate of host in g/s" + query: uuid:ceems_host_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} + kind: range # GPU utilization - avg_gpu_usage: uuid:ceems_gpu_usage:ratio{uuid=~"%s"} + - name: avg_gpu_usage + title: "GPU Usage (%)" + help: "Usage of GPU in %" + query: uuid:ceems_gpu_usage:ratio{uuid=~`{{.UUIDs}}`} + kind: range # GPU memory utilization - avg_gpu_mem_usage: uuid:ceems_gpu_memory_usage:ratio{uuid=~"%s"} + - name: avg_gpu_mem_usage + title: "GPU Memory Usage (%)" + help: "Ratio of memory used to memory reserved of GPU in %" + query: uuid:ceems_gpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} + kind: range # GPU power usage in Watts - gpu_power_usage: uuid:ceems_gpu_power_watts:pue{uuid=~"%s"} + - name: gpu_power_usage + title: "GPU Power Usage (W)" + help: "Instanteous Power Usage of GPU in Watts" + query: uuid:ceems_gpu_power_watts:pue{uuid=~`{{.UUIDs}}`} + kind: range # GPU emissions in g/s - gpu_emissions: uuid:ceems_gpu_emissions_g_s:pue{uuid=~"%s"} + - name: gpu_emissions + title: "GPU Eq. Emission Rate (g/s)" + help: "Instanteous emissions rate of GPU in g/s" + query: uuid:ceems_gpu_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} + kind: range # Read IO bytes/s - io_read_bytes: irate(ceems_ebpf_read_bytes_total{uuid=~"%s"}[1m]) + - name: io_read_bytes + title: "IO Read Bandwidth (b/s)" + help: "Instanteous IO read bandwidth in bytes/s" + query: irate(ceems_ebpf_read_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) + kind: range # Write IO bytes/s - io_write_bytes: irate(ceems_ebpf_write_bytes_total{uuid=~"%s"}[1m]) + - name: io_write_bytes + title: "IO Write Bandwidth (b/s)" + help: "Instanteous IO write bandwidth in bytes/s" + query: irate(ceems_ebpf_write_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) + kind: range ``` A complete reference can be found in the [Reference](./config-reference.md) section. A valid diff --git a/website/docs/configuration/config-reference.md b/website/docs/configuration/config-reference.md index ae942699..21bf6d01 100644 --- a/website/docs/configuration/config-reference.md +++ b/website/docs/configuration/config-reference.md @@ -232,7 +232,7 @@ basic_auth: # authorization: # Sets the authentication type of the request. - [ type: | default: Bearer ] + [ type: | default = Bearer ] # Sets the credentials of the request. It is mutually exclusive with # `credentials_file`. [ credentials: ] @@ -250,7 +250,7 @@ oauth2: [ follow_redirects: | default = true ] # Whether to enable HTTP2. -[ enable_http2: | default: true ] +[ enable_http2: | default = true ] # Configures the API request's TLS settings. # @@ -510,7 +510,7 @@ extra_config: # # Default value is 50 # - [ query_max_series: | default: 50 ] + [ query_max_series: | default = 50 ] # Minimum number of samples that are guaranteed to be available for executing the queries # of the updater. It is expressed as a proportion of `--query.max-samples` and takes a value @@ -518,7 +518,7 @@ extra_config: # # Default value is 0.5 # - [ query_min_samples: | default: 0.5 ] + [ query_min_samples: | default = 0.5 ] # Scrape interval corresponding to the scrape targets that generate metrics provided in # `queries` section. @@ -527,7 +527,7 @@ extra_config: # # Units Supported: y, w, d, h, m, s, ms. # - [ scrape_interval: | default: 0s ] + [ scrape_interval: | default = 0s ] # Evaluation interval corresponding to the recording rules that generate metrics provided in # `queries` section. @@ -536,7 +536,7 @@ extra_config: # # Units Supported: y, w, d, h, m, s, ms. # - [ evaluation_interval: | default: 0s ] + [ evaluation_interval: | default = 0s ] # Compute units that have a total lifetime less than this value will be deleted from # TSDB to reduce the number of labels and cardinality. @@ -545,14 +545,14 @@ extra_config: # # Units Supported: y, w, d, h, m, s, ms. # - [ cutoff_duration: | default: 0s ] + [ cutoff_duration: | default = 0s ] # The ignored units' (based on `cutoff_duration`) metrics will be dropped from the TSDB # when set to `true`. This can be used to reduce the number of labels and cardinality of TSDB. # # TSDB must be started with the `--web.enable-admin-api` flag for this to work. # - [ delete_ignored: | default: false ] + [ delete_ignored: | default = false ] # List of labels to delete from TSDB. These labels should be valid matchers for TSDB. # More information on the delete API of Prometheus: https://prometheus.io/docs/prometheus/latest/querying/api/#delete-series @@ -564,15 +564,6 @@ extra_config: # Define queries that are used to estimate aggregate metrics of each compute unit. # These queries will be passed to golang's text/template package to build them. - # Available template variables: - # - UUIDs -> UUIDs string delimited by "|", e.g., 123|345|567 - # - ScrapeInterval -> Scrape interval of TSDB in time.Duration format, e.g., 15s, 1m - # - ScrapeIntervalMilli -> Scrape interval of TSDB in milliseconds, e.g., 15000, 60000 - # - EvaluationInterval -> Evaluation interval of TSDB in time.Duration format, e.g., 15s, 1m - # - EvaluationIntervalMilli -> Evaluation interval of TSDB in milliseconds, e.g., 15s, 1m - # - RateInterval -> Rate interval in time.Duration format. It is estimated based on Scrape interval as 4*scrape_interval - # - Range -> Duration of interval where aggregation is being made in time.Duration format - # queries: [ ] ``` @@ -608,6 +599,18 @@ A `queries_config` allows configuring PromQL queries for the TSDB updater of the # metrics. If operators deploy more exporters of their own, queries # must be modified accordingly. # +# Available template variables: +# - UUIDs -> UUIDs string delimited by "|", e.g., 123|345|567 +# - ScrapeInterval -> Scrape interval of TSDB in time.Duration format, e.g., 15s, 1m +# - ScrapeIntervalMilli -> Scrape interval of TSDB in milliseconds, e.g., 15000, 60000 +# - EvaluationInterval -> Evaluation interval of TSDB in time.Duration format, e.g., 15s, 1m +# - EvaluationIntervalMilli -> Evaluation interval of TSDB in milliseconds, e.g., 15s, 1m +# - RateInterval -> Rate interval in time.Duration format. It is estimated based on Scrape interval as 4*scrape_interval +# - Range -> Duration of interval where aggregation is being made in time.Duration format +# +# IMPORTANT: Always use backticks around {{.UUIDs}} template variable to escape characters +# like "[" and "]" which can be found in identifiers of resource managers like LSF. +# # Average CPU utilization # # Default value: @@ -913,29 +916,35 @@ A valid sample configuration file can be found in the # logging: # Enable system logging - enabled: false + [ enabled: | default = false] # Logging level. Valid options are + # # - info # - debug # - warn # - error # - level: info + # Default: info + # + [ level: | default = info ] # Logging format. Valid options are + # # - logfmt # - json # - warn # - error # - format: logfmt + # Default: logfmt + # + [ format: | default = logfmt ] # Directory where logging file will be saved. # This directory must exist with correct # permissions for logging file to be created. # - directory: /var/log/ceems + [ directory: | default = /var/log/ceems ] # Configuration of the CEEMS API server # @@ -964,46 +973,129 @@ tsdb: # web: + # Max number of units' time series data to be fetched in a single CLI execution. Use + # an appropriate number to avoid spiking the memory usage of TSDB server. + # + # Default: 10 + # + [ max_units: | default = 10 ] + + # Scrape interval corresponding to the scrape targets that generate metrics provided in + # `queries` section. + # + # Default value `0s` means global scrape interval of the TSDB instance will be used. + # + # Units Supported: y, w, d, h, m, s, ms. + # + [ scrape_interval: | default = 0s ] + + # Evaluation interval corresponding to the recording rules that generate metrics provided in + # `queries` section. + # + # Default value `0s` means global evaluation interval of the TSDB instance will be used. + # + # Units Supported: y, w, d, h, m, s, ms. + # + [ evaluation_interval: | default = 0s ] + # To dump the time series data for each metric, this section must be configured. # The key name is the name of the metric, and the value is the PromQL query to get - # time series data. The placeholder `%s` will be replaced by a list of job IDs delimited - # by `|`, which is the syntax expected by the TSDB server. + # time series data. # # If the TSDB server has been configured with the recording rules generated by `ceems_tool`, # the following queries should work out-of-the-box. # - # # CPU utilization - # cpu_usage: uuid:ceems_cpu_usage:ratio_irate{uuid=~"%s"} - + # CPU utilization + # - name: cpu_usage + # title: "CPU Usage (%)" + # help: "Usage of CPUs in %" + # query: uuid:ceems_cpu_usage:ratio_irate{uuid=~`{{.UUIDs}}`} + # # # CPU Memory utilization - # cpu_mem_usage: uuid:ceems_cpu_memory_usage:ratio{uuid=~"%s"} - + # - name: cpu_mem_usage + # title: "CPU Memory Usage (%)" + # help: "Ratio of memory used to memory reserved of CPU in %" + # query: uuid:ceems_cpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} + # # # Host power usage in Watts - # host_power_usage: uuid:ceems_host_power_watts:pue{uuid=~"%s"} - + # - name: host_power_usage + # title: "Host Power usage (W)" + # help: "Instanteous power usage of host in Watts" + # query: uuid:ceems_host_power_watts:pue{uuid=~`{{.UUIDs}}`} + # # # Host emissions in g/s - # host_emissions: uuid:ceems_host_emissions_g_s:pue{uuid=~"%s"} - + # - name: host_emissions + # title: "Host Eq. Emissions Rate (g/s)" + # help: "Instanteous emissions rate of host in g/s" + # query: uuid:ceems_host_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} + # # # GPU utilization - # avg_gpu_usage: uuid:ceems_gpu_usage:ratio{uuid=~"%s"} - + # - name: avg_gpu_usage + # title: "GPU Usage (%)" + # help: "Usage of GPU in %" + # query: uuid:ceems_gpu_usage:ratio{uuid=~`{{.UUIDs}}`} + # # # GPU memory utilization - # avg_gpu_mem_usage: uuid:ceems_gpu_memory_usage:ratio{uuid=~"%s"} - + # - name: avg_gpu_mem_usage + # title: "GPU Memory Usage (%)" + # help: "Ratio of memory used to memory reserved of GPU in %" + # query: uuid:ceems_gpu_memory_usage:ratio{uuid=~`{{.UUIDs}}`} + # # # GPU power usage in Watts - # gpu_power_usage: uuid:ceems_gpu_power_watts:pue{uuid=~"%s"} - + # - name: gpu_power_usage + # title: "GPU Power Usage (W)" + # help: "Instanteous Power Usage of GPU in Watts" + # query: uuid:ceems_gpu_power_watts:pue{uuid=~`{{.UUIDs}}`} + # # # GPU emissions in g/s - # gpu_emissions: uuid:ceems_gpu_emissions_g_s:pue{uuid=~"%s"} + # - name: gpu_emissions + # title: "GPU Eq. Emission Rate (g/s)" + # help: "Instanteous emissions rate of GPU in g/s" + # query: uuid:ceems_gpu_emissions_g_s:pue{uuid=~`{{.UUIDs}}`} + # + # # Read IO bytes/s + # - name: io_read_bytes + # title: "IO Read Bandwidth (b/s)" + # help: "Instanteous IO read bandwidth in bytes/s" + # query: irate(ceems_ebpf_read_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) + # + # # Write IO bytes/s + # - name: io_write_bytes + # title: "IO Write Bandwidth (b/s)" + # help: "Instanteous IO write bandwidth in bytes/s" + # query: irate(ceems_ebpf_write_bytes_total{uuid=~`{{.UUIDs}}`}[1m]) + range_queries: + [ - ] +``` - # # Read IO bytes - # io_read_bytes: irate(ceems_ebpf_read_bytes_total{uuid=~"%s"}[1m]) +### `` - # # Write IO bytes - # io_write_bytes: irate(ceems_ebpf_write_bytes_total{uuid=~"%s"}[1m]) +A `tsdb_query` allows to configure TSDB metrics in the `cacct` config file. + +```yaml +tsdb_query: + # A UNIQUE short name for the query + [ name: ] + # A human readable name for the query + [ title: ] + # A small help text explaining the query metrics + [ help: ] + # Valid PromQL query # - queries: - [ : ... ] + # Available template variables: + # + # - UUIDs -> UUIDs string delimited by "|", e.g., 123|345|567 + # - ScrapeInterval -> Scrape interval of TSDB in time.Duration format, e.g., 15s, 1m + # - ScrapeIntervalMilli -> Scrape interval of TSDB in milliseconds, e.g., 15000, 60000 + # - EvaluationInterval -> Evaluation interval of TSDB in time.Duration format, e.g., 15s, 1m + # - EvaluationIntervalMilli -> Evaluation interval of TSDB in milliseconds, e.g., 15s, 1m + # - RateInterval -> Rate interval in time.Duration format. It is estimated based on Scrape interval as 4*scrape_interval + # - Range -> Duration of interval where aggregation is being made in time.Duration format + # + # IMPORTANT: Always use backticks around {{.UUIDs}} template variable to escape characters + # like "[" and "]" which can be found in identifiers of resource managers like LSF. + # + [ query: ] ``` ## `` @@ -1029,7 +1121,7 @@ basic_auth: # authorization: # Sets the authentication type of the request. - [ type: | default: Bearer ] + [ type: | default = Bearer ] # Sets the credentials of the request. It is mutually exclusive with # `credentials_file`. [ credentials: ] @@ -1047,7 +1139,7 @@ oauth2: [ follow_redirects: | default = true ] # Whether to enable HTTP2. -[ enable_http2: | default: true ] +[ enable_http2: | default = true ] # Configures the API request's TLS settings. # @@ -1095,7 +1187,7 @@ tls_config: # contain port numbers. [ no_proxy: ] # Use proxy URL indicated by environment variables (HTTP_PROXY, https_proxy, HTTPs_PROXY, https_proxy, and no_proxy) -[ proxy_from_environment: | default: false ] +[ proxy_from_environment: | default = false ] # Specifies headers to send to proxies during CONNECT requests. [ proxy_connect_header: [ : [, ...] ] ] diff --git a/website/docs/usage/cacct.md b/website/docs/usage/cacct.md index 2f618226..a93d1e16 100644 --- a/website/docs/usage/cacct.md +++ b/website/docs/usage/cacct.md @@ -73,14 +73,14 @@ data, the `--ts` flag must be passed. :::important[IMPORTANT] -When the `--ts` flag is used, it is compulsory to set at least one compute unit ID using +When the `--ts.metrics` flag is set, it is compulsory to set at least one compute unit ID using the `--job` flag. If users want time series data for multiple jobs, a comma-separated list of IDs can be passed to the `--job` flag. ::: ```bash -cacct --job=1234,1233 --ts --ts.out-dir=data +cacct --job=1234,1233 --ts.metrics="cpuusage" --ts.out-dir=data ``` With the above command, the time series data of compute units 1234 and 1233 will be saved @@ -94,11 +94,8 @@ a typical `metadata.json` would be as follows: { "fingerprint": "d2213312c639a90c", "labels": { - "__name__": "uuid:ceems_host_emissions_g_s:pue", - "hostname": "ceems-demo", - "instance": "localhost:9010", - "job": "slurm", - "manager": "slurm", + "metric": "uuid:ceems_host_emissions_g_s:pue", + "nodename": "ceems-demo", "provider": "owid", "uuid": "258" } @@ -106,11 +103,8 @@ a typical `metadata.json` would be as follows: { "fingerprint": "85105ad7ffcf540a", "labels": { - "__name__": "uuid:ceems_host_emissions_g_s:pue", - "hostname": "ceems-demo", - "instance": "localhost:9010", - "job": "slurm", - "manager": "slurm", + "metric": "uuid:ceems_host_emissions_g_s:pue", + "nodename": "ceems-demo", "provider": "rte", "uuid": "258" } @@ -118,33 +112,24 @@ a typical `metadata.json` would be as follows: { "fingerprint": "c819bde6e9a529b6", "labels": { - "__name__": "uuid:ceems_cpu_memory_usage:ratio", - "hostname": "ceems-demo", - "instance": "localhost:9010", - "job": "slurm", - "manager": "slurm", + "metric": "uuid:ceems_cpu_memory_usage:ratio", + "nodename": "ceems-demo", "uuid": "258" } }, { "fingerprint": "90bcc7cfa3cd05fa", "labels": { - "__name__": "uuid:ceems_cpu_usage:ratio_irate", - "hostname": "ceems-demo", - "instance": "localhost:9010", - "job": "slurm", - "manager": "slurm", + "metric": "uuid:ceems_cpu_usage:ratio_irate", + "nodename": "ceems-demo", "uuid": "258" } }, { "fingerprint": "cbba6b4919ac1bad", "labels": { - "__name__": "uuid:ceems_host_power_watts:pue", - "hostname": "ceems-demo", - "instance": "localhost:9010", - "job": "slurm", - "manager": "slurm", + "metric": "uuid:ceems_host_power_watts:pue", + "nodename": "ceems-demo", "uuid": "258" } } diff --git a/website/md-link-check.json b/website/md-link-check.json index fdd75539..6fab8476 100644 --- a/website/md-link-check.json +++ b/website/md-link-check.json @@ -30,7 +30,10 @@ }, { "pattern": "https://www.supermicro.com/en/glossary/baseboard-management-controller" - } + }, + { + "pattern": "https://www.ibm.com" + } ], "replacementPatterns": [ {