diff --git a/components/usage/pkg/scheduler/job.go b/components/usage/pkg/scheduler/job.go index ac57b69d92817f..8ace18129d1605 100644 --- a/components/usage/pkg/scheduler/job.go +++ b/components/usage/pkg/scheduler/job.go @@ -5,12 +5,9 @@ package scheduler import ( - "context" "fmt" "github.com/gitpod-io/gitpod/common-go/log" - v1 "github.com/gitpod-io/gitpod/usage-api/v1" "github.com/robfig/cron" - "google.golang.org/protobuf/types/known/timestamppb" "time" ) @@ -37,51 +34,6 @@ func NewPeriodicJobSpec(period time.Duration, id string, job Job) (JobSpec, erro }, nil } -func NewLedgerTriggerJobSpec(schedule time.Duration, job Job) (JobSpec, error) { - return NewPeriodicJobSpec(schedule, "ledger", WithoutConcurrentRun(job)) -} - -func NewLedgerTrigger(usageClient v1.UsageServiceClient, billingClient v1.BillingServiceClient) *LedgerJob { - return &LedgerJob{ - usageClient: usageClient, - billingClient: billingClient, - } -} - -type LedgerJob struct { - usageClient v1.UsageServiceClient - billingClient v1.BillingServiceClient -} - -func (r *LedgerJob) Run() (err error) { - ctx := context.Background() - now := time.Now().UTC() - hourAgo := now.Add(-1 * time.Hour) - - logger := log. - WithField("from", hourAgo). - WithField("to", now) - - logger.Info("Running ledger job. Reconciling usage records.") - _, err = r.usageClient.ReconcileUsage(ctx, &v1.ReconcileUsageRequest{ - From: timestamppb.New(hourAgo), - To: timestamppb.New(now), - }) - if err != nil { - logger.WithError(err).Errorf("Failed to reconcile usage with ledger.") - return fmt.Errorf("failed to reconcile usage with ledger: %w", err) - } - - logger.Info("Starting invoice reconciliation.") - _, err = r.billingClient.ReconcileInvoices(ctx, &v1.ReconcileInvoicesRequest{}) - if err != nil { - logger.WithError(err).Errorf("Failed to reconcile invoices.") - return fmt.Errorf("failed to reconcile invoices: %w", err) - } - - return nil -} - // WithoutConcurrentRun wraps a Job and ensures the job does not concurrently func WithoutConcurrentRun(j Job) Job { return &preventConcurrentInvocation{ diff --git a/components/usage/pkg/scheduler/ledger_job.go b/components/usage/pkg/scheduler/ledger_job.go new file mode 100644 index 00000000000000..d7554ad3e94558 --- /dev/null +++ b/components/usage/pkg/scheduler/ledger_job.go @@ -0,0 +1,63 @@ +// Copyright (c) 2022 Gitpod GmbH. All rights reserved. +// Licensed under the GNU Affero General Public License (AGPL). +// See License-AGPL.txt in the project root for license information. + +package scheduler + +import ( + "context" + "fmt" + "github.com/gitpod-io/gitpod/common-go/log" + v1 "github.com/gitpod-io/gitpod/usage-api/v1" + "google.golang.org/protobuf/types/known/timestamppb" + "time" +) + +func NewLedgerTriggerJobSpec(schedule time.Duration, job Job) (JobSpec, error) { + return NewPeriodicJobSpec(schedule, "ledger", WithoutConcurrentRun(job)) +} + +func NewLedgerTrigger(usageClient v1.UsageServiceClient, billingClient v1.BillingServiceClient) *LedgerJob { + return &LedgerJob{ + usageClient: usageClient, + billingClient: billingClient, + } +} + +type LedgerJob struct { + usageClient v1.UsageServiceClient + billingClient v1.BillingServiceClient +} + +func (r *LedgerJob) Run() (err error) { + defer func() { + reportLedgerCompleted(err) + }() + + ctx := context.Background() + now := time.Now().UTC() + hourAgo := now.Add(-1 * time.Hour) + + logger := log. + WithField("from", hourAgo). + WithField("to", now) + + logger.Info("Running ledger job. Reconciling usage records.") + _, err = r.usageClient.ReconcileUsage(ctx, &v1.ReconcileUsageRequest{ + From: timestamppb.New(hourAgo), + To: timestamppb.New(now), + }) + if err != nil { + logger.WithError(err).Errorf("Failed to reconcile usage with ledger.") + return fmt.Errorf("failed to reconcile usage with ledger: %w", err) + } + + logger.Info("Starting invoice reconciliation.") + _, err = r.billingClient.ReconcileInvoices(ctx, &v1.ReconcileInvoicesRequest{}) + if err != nil { + logger.WithError(err).Errorf("Failed to reconcile invoices.") + return fmt.Errorf("failed to reconcile invoices: %w", err) + } + + return nil +} diff --git a/components/usage/pkg/scheduler/reporter.go b/components/usage/pkg/scheduler/reporter.go index 7020885b6fd1ef..a10fb73f7d4d96 100644 --- a/components/usage/pkg/scheduler/reporter.go +++ b/components/usage/pkg/scheduler/reporter.go @@ -31,6 +31,14 @@ var ( Buckets: prometheus.LinearBuckets(30, 30, 10), // every 30 secs, starting at 30secs }, []string{"job", "outcome"}) + ledgerLastCompletedTime = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "ledger_last_completed_time", + Help: "The last time the ledger scheduled job completed, by outcome", + ConstLabels: nil, + }, []string{"outcome"}) + stoppedWithoutStoppingTime = prometheus.NewGauge(prometheus.GaugeOpts{ Namespace: namespace, Subsystem: subsystem, @@ -44,6 +52,7 @@ func RegisterMetrics(reg *prometheus.Registry) error { jobStartedSeconds, jobCompletedSeconds, stoppedWithoutStoppingTime, + ledgerLastCompletedTime, } for _, metric := range metrics { err := reg.Register(metric) @@ -60,9 +69,17 @@ func reportJobStarted(id string) { } func reportJobCompleted(id string, duration time.Duration, err error) { - outcome := "success" + jobCompletedSeconds.WithLabelValues(id, outcomeFromErr(err)).Observe(duration.Seconds()) +} + +func reportLedgerCompleted(err error) { + ledgerLastCompletedTime.WithLabelValues(outcomeFromErr(err)).SetToCurrentTime() +} + +func outcomeFromErr(err error) string { + out := "success" if err != nil { - outcome = "error" + out = "error" } - jobCompletedSeconds.WithLabelValues(id, outcome).Observe(duration.Seconds()) + return out }