Merge pull request #2602 from ameijer/atm/plugin-registration-cp

core/pkg/model/customcostresponse.go

@@ -134,6 +134,95 @@ type ExtendedCustomCostAttributes struct {
 	PricingCategory string
 }
 
+func (c *CustomCostResponse) Clone() CustomCostResponse {
+	win := c.GetWindow().Clone()
+	costClones := []*CustomCost{}
+
+	for _, cost := range c.GetCosts() {
+		clone := cost.Clone()
+		costClones = append(costClones, &clone)
+	}
+
+	errClones := []error{}
+	for _, err := range c.GetErrors() {
+		errClones = append(errClones, err)
+	}
+	return CustomCostResponse{
+		Metadata:   cloneMap(c.GetMetadata()),
+		Costsource: c.GetCostSource(),
+		Domain:     c.GetDomain(),
+		Version:    c.GetVersion(),
+		Currency:   c.GetCurrency(),
+		Window:     win,
+		Costs:      costClones,
+		Errors:     errClones,
+	}
+}
+
+func cloneMap(input map[string]string) map[string]string {
+	if input == nil {
+		return nil
+	}
+	result := map[string]string{}
+	for key, val := range input {
+		result[key] = val
+	}
+	return result
+}
+
+func (c *CustomCost) Clone() CustomCost {
+	win := c.GetWindow().Clone()
+	var ext ExtendedCustomCostAttributes
+	if c.GetExtendedAttributes() != nil {
+		ext = c.GetExtendedAttributes().Clone()
+	}
+	return CustomCost{
+		Metadata:           cloneMap(c.GetMetadata()),
+		Zone:               c.GetCostIncurredZone(),
+		BilledCost:         c.GetBilledCost(),
+		AccountName:        c.GetAccountName(),
+		ChargeCategory:     c.GetChargeCategory(),
+		Description:        c.GetDescription(),
+		ListCost:           c.GetListCost(),
+		ListUnitPrice:      c.GetListUnitPrice(),
+		ResourceName:       c.GetResourceName(),
+		ResourceType:       c.GetResourceType(),
+		Id:                 c.GetID(),
+		ProviderId:         c.GetProviderID(),
+		Window:             &win,
+		Labels:             cloneMap(c.GetLabels()),
+		UsageQty:           c.GetUsageQuantity(),
+		UsageUnit:          c.GetUsageUnit(),
+		ExtendedAttributes: &ext,
+	}
+}
+func (e *ExtendedCustomCostAttributes) Clone() ExtendedCustomCostAttributes {
+	win := e.BillingPeriod.Clone()
+	return ExtendedCustomCostAttributes{
+		BillingPeriod:              &win,
+		AccountID:                  e.GetAccountID(),
+		ChargeFrequency:            e.GetChargeFrequency(),
+		Subcategory:                e.GetSubcategory(),
+		CommitmentDiscountCategory: e.GetCommitmentDiscountCategory(),
+		CommitmentDiscountID:       e.GetCommitmentDiscountID(),
+		CommitmentDiscountName:     e.GetCommitmentDiscountName(),
+		CommitmentDiscountType:     e.GetCommitmentDiscountType(),
+		EffectiveCost:              e.GetEffectiveCost(),
+		InvoiceIssuer:              e.GetInvoiceIssuer(),
+		Provider:                   e.GetProvider(),
+		Publisher:                  e.GetPublisher(),
+		ServiceCategory:            e.GetServiceCategory(),
+		ServiceName:                e.GetServiceName(),
+		SkuID:                      e.GetSKUID(),
+		SkuPriceID:                 e.GetSKUPriceID(),
+		SubAccountID:               e.GetSubAccountID(),
+		SubAccountName:             e.GetSubAccountName(),
+		PricingQuantity:            e.GetPricingQuantity(),
+		PricingUnit:                e.GetPricingUnit(),
+		PricingCategory:            e.GetPricingCategory(),
+	}
+}
+
 func (e *ExtendedCustomCostAttributes) GetBillingPeriod() *opencost.Window {
 	return e.BillingPeriod
 }

core/pkg/version/version.go

@@ -3,8 +3,9 @@ package version
 import "fmt"
 
 var (
-	Version   = "dev"
-	GitCommit = "HEAD"
+	Version      = "dev"
+	GitCommit    = "HEAD"
+	Architecture = "amd64"
 )
 
 func FriendlyVersion() string {

justfile

@@ -26,6 +26,7 @@ build-binary VERSION=version:
         {{commonenv}} GOOS=linux GOARCH=amd64 go build \
         -ldflags \
           "-X github.com/opencost/opencost/pkg/version.Version={{VERSION}} \
+           -X github.com/opencost/opencost/pkg/version.Architecture=amd64 \
            -X github.com/opencost/opencost/pkg/version.GitCommit={{commit}}" \
         -o ./costmodel-amd64
 
@@ -33,6 +34,7 @@ build-binary VERSION=version:
         {{commonenv}} GOOS=linux GOARCH=arm64 go build \
         -ldflags \
           "-X github.com/opencost/opencost/pkg/version.Version={{VERSION}} \
+           -X github.com/opencost/opencost/pkg/version.Architecture=arm64 \
            -X github.com/opencost/opencost/pkg/version.GitCommit={{commit}}" \
         -o ./costmodel-arm64
 

pkg/cmd/costmodel/costmodel.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/julienschmidt/httprouter"
 	"github.com/opencost/opencost/pkg/cloudcost"
+	"github.com/opencost/opencost/pkg/customcost"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/cors"
 
@@ -57,6 +58,20 @@ func Execute(opts *CostModelOpts) error {
 		a.CloudCostQueryService = cloudcost.NewQueryService(repoQuerier, repoQuerier)
 	}
 
+	log.Infof("Custom Costs enabled: %t", env.IsCustomCostEnabled())
+	if env.IsCustomCostEnabled() {
+		hourlyRepo := customcost.NewMemoryRepository()
+		dailyRepo := customcost.NewMemoryRepository()
+		ingConfig := customcost.DefaultIngestorConfiguration()
+		var err error
+		a.CustomCostPipelineService, err = customcost.NewPipelineService(hourlyRepo, dailyRepo, ingConfig)
+		if err != nil {
+			return fmt.Errorf("error instantiating custom cost pipeline service: %v", err)
+		}
+		//repoQuerier := cloudcost.NewRepositoryQuerier(repo)
+		//a.CloudCostQueryService = cloudcost.NewQueryService(repoQuerier, repoQuerier)
+	}
+
 	rootMux := http.NewServeMux()
 	a.Router.GET("/healthz", Healthz)
 

pkg/costmodel/router.go

@@ -28,11 +28,11 @@ import (
 	"github.com/opencost/opencost/pkg/cloudcost"
 	"github.com/opencost/opencost/pkg/config"
 	clustermap "github.com/opencost/opencost/pkg/costmodel/clusters"
+	"github.com/opencost/opencost/pkg/customcost"
 	"github.com/opencost/opencost/pkg/kubeconfig"
 	"github.com/opencost/opencost/pkg/metrics"
 	"github.com/opencost/opencost/pkg/services"
 	"github.com/spf13/viper"
-
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/julienschmidt/httprouter"
@@ -86,26 +86,27 @@ var (
 // Accesses defines a singleton application instance, providing access to
 // Prometheus, Kubernetes, the cloud provider, and caches.
 type Accesses struct {
-	Router                   *httprouter.Router
-	PrometheusClient         prometheus.Client
-	ThanosClient             prometheus.Client
-	KubeClientSet            kubernetes.Interface
-	ClusterCache             clustercache.ClusterCache
-	ClusterMap               clusters.ClusterMap
-	CloudProvider            models.Provider
-	ConfigFileManager        *config.ConfigFileManager
-	CloudConfigController    *cloudconfig.Controller
-	CloudCostPipelineService *cloudcost.PipelineService
-	CloudCostQueryService    *cloudcost.QueryService
-	ClusterInfoProvider      clusters.ClusterInfoProvider
-	Model                    *CostModel
-	MetricsEmitter           *CostModelMetricsEmitter
-	OutOfClusterCache        *cache.Cache
-	AggregateCache           *cache.Cache
-	CostDataCache            *cache.Cache
-	ClusterCostsCache        *cache.Cache
-	CacheExpiration          map[time.Duration]time.Duration
-	AggAPI                   Aggregator
+	Router                    *httprouter.Router
+	PrometheusClient          prometheus.Client
+	ThanosClient              prometheus.Client
+	KubeClientSet             kubernetes.Interface
+	ClusterCache              clustercache.ClusterCache
+	ClusterMap                clusters.ClusterMap
+	CloudProvider             models.Provider
+	ConfigFileManager         *config.ConfigFileManager
+	CloudConfigController     *cloudconfig.Controller
+	CloudCostPipelineService  *cloudcost.PipelineService
+	CloudCostQueryService     *cloudcost.QueryService
+	CustomCostPipelineService *customcost.PipelineService
+	ClusterInfoProvider       clusters.ClusterInfoProvider
+	Model                     *CostModel
+	MetricsEmitter            *CostModelMetricsEmitter
+	OutOfClusterCache         *cache.Cache
+	AggregateCache            *cache.Cache
+	CostDataCache             *cache.Cache
+	ClusterCostsCache         *cache.Cache
+	CacheExpiration           map[time.Duration]time.Duration
+	AggAPI                    Aggregator
 	// SettingsCache stores current state of app settings
 	SettingsCache *cache.Cache
 	// settingsSubscribers tracks channels through which changes to different

pkg/customcost/ingestor.go

@@ -0,0 +1,572 @@
+package customcost
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/davecgh/go-spew/spew"
+	"github.com/hashicorp/go-plugin"
+	"github.com/opencost/opencost/core/pkg/log"
+	"github.com/opencost/opencost/core/pkg/model"
+	"github.com/opencost/opencost/core/pkg/opencost"
+	"github.com/opencost/opencost/core/pkg/util/timeutil"
+	"github.com/opencost/opencost/pkg/env"
+)
+
+// IngestorStatus includes diagnostic values for a given Ingestor
+type IngestorStatus struct {
+	Created  time.Time
+	LastRun  time.Time
+	NextRun  time.Time
+	Runs     int
+	Coverage opencost.Window
+}
+
+// CustomCost IngestorConfig is a configuration struct for an Ingestor
+type CustomCostIngestorConfig struct {
+	MonthToDateRunInterval               int
+	HourlyDuration, DailyDuration        time.Duration
+	QueryWindow                          time.Duration
+	RunWindow                            time.Duration
+	PluginConfigDir, PluginExecutableDir string
+	RefreshRate                          time.Duration
+}
+
+// DefaultIngestorConfiguration retrieves an CustomCostIngestorConfig from env variables
+func DefaultIngestorConfiguration() CustomCostIngestorConfig {
+	return CustomCostIngestorConfig{
+		DailyDuration:          timeutil.Day * time.Duration(env.GetDataRetentionDailyResolutionDays()),
+		HourlyDuration:         time.Hour * time.Duration(env.GetDataRetentionHourlyResolutionHours()),
+		MonthToDateRunInterval: env.GetCloudCostMonthToDateInterval(),
+		QueryWindow:            timeutil.Day * time.Duration(env.GetCloudCostQueryWindowDays()),
+		PluginConfigDir:        env.GetPluginConfigDir(),
+	}
+}
+
+type CustomCostIngestor struct {
+	key          string
+	config       *CustomCostIngestorConfig
+	repo         Repository
+	runID        string
+	lastRun      time.Time
+	runs         int
+	creationTime time.Time
+	coverage     opencost.Window
+	coverageLock sync.Mutex
+	isRunning    atomic.Bool
+	isStopping   atomic.Bool
+	exitBuildCh  chan string
+	exitRunCh    chan string
+}
+
+// NewIngestor is an initializer for ingestor
+func NewCustomCostIngestor(ingestorConfig *CustomCostIngestorConfig, repo Repository, plugins map[string]*plugin.Client) (*CustomCostIngestor, error) {
+	if repo == nil {
+		return nil, fmt.Errorf("CustomCost: NewCustomCostIngestor: repository connot be nil")
+	}
+	if ingestorConfig == nil {
+		return nil, fmt.Errorf("CustomCost: NewCustomCostIngestor: integration connot be nil")
+	}
+
+	now := time.Now().UTC()
+	midnight := opencost.RoundForward(now, timeutil.Day)
+	return &CustomCostIngestor{
+		config:       ingestorConfig,
+		repo:         repo,
+		creationTime: now,
+		lastRun:      now,
+		coverage:     opencost.NewClosedWindow(midnight, midnight),
+	}, nil
+}
+
+func (ing *CustomCostIngestor) LoadWindow(start, end time.Time) {
+	windows, err := opencost.GetWindows(start, end, timeutil.Day)
+	if err != nil {
+		log.Errorf("CloudCost[%s]: ingestor: invalid window %s", ing.key, opencost.NewWindow(&start, &end))
+		return
+	}
+
+	for _, window := range windows {
+		has, err2 := ing.repo.Has(*window.Start(), ing.key)
+		if err2 != nil {
+			log.Errorf("CloudCost[%s]: ingestor: error when loading window: %s", ing.key, err2.Error())
+		}
+		if !has {
+			ing.BuildWindow(start, end)
+			return
+		}
+		ing.expandCoverage(window)
+		log.Debugf("CloudCost[%s]: ingestor: skipping build for window %s, coverage already exists", ing.key, window.String())
+	}
+
+}
+
+func (ing *CustomCostIngestor) BuildWindow(start, end time.Time) {
+	// log.Infof("CloudCost[%s]: ingestor: building window %s", ing.key, opencost.NewWindow(&start, &end))
+	// ccsr, err := ing.integration.GetCloudCost(start, end)
+	// if err != nil {
+	// 	log.Errorf("CloudCost[%s]: ingestor: build failed for window %s: %s", ing.key, opencost.NewWindow(&start, &end), err.Error())
+	// 	return
+	// }
+	// for _, ccs := range ccsr.CloudCostSets {
+	// 	log.Debugf("BuildWindow[%s]: GetCloudCost: writing cloud costs for window %s: %d", ccs.Integration, ccs.Window, len(ccs.CloudCosts))
+	// 	err2 := ing.repo.Put(ccs)
+	// 	if err2 != nil {
+	// 		log.Errorf("CloudCost[%s]: ingestor: failed to save Cloud Cost Set with window %s: %s", ing.key, ccs.GetWindow().String(), err2.Error())
+	// 	}
+	// 	ing.expandCoverage(ccs.Window)
+	// }
+}
+
+func (ing *CustomCostIngestor) Start(rebuild bool) {
+
+	// // If already running, log that and return.
+	// if !ing.isRunning.CompareAndSwap(false, true) {
+	// 	log.Infof("CloudCost: ingestor: is already running")
+	// 	return
+	// }
+
+	// ing.runID = stringutil.RandSeq(5)
+
+	// ing.exitBuildCh = make(chan string)
+	// ing.exitRunCh = make(chan string)
+
+	// // Build the store once, advancing backward in time from the earliest
+	// // point of coverage.
+	// go ing.build(rebuild)
+
+	// go ing.run()
+
+	// TEMPORARY - load the repo with dummy data
+	var resps []*model.CustomCostResponse
+	err := json.Unmarshal([]byte(ddData), &resps)
+	if err != nil {
+		panic(err)
+	}
+	err = ing.repo.Put(resps)
+	if err != nil {
+		panic(err)
+	}
+	//2024-02-27T01:00:00
+	target := time.Date(2024, 2, 27, 1, 0, 0, 0, time.UTC)
+	stored, err := ing.repo.Get(target, "datadog")
+	if err != nil {
+		panic(err)
+	}
+
+	for _, storedResp := range stored {
+		log.Debug("got stored object: ")
+		spew.Dump(storedResp)
+	}
+}
+
+func (ing *CustomCostIngestor) Stop() {
+	// If already stopping, log that and return.
+	if !ing.isStopping.CompareAndSwap(false, true) {
+		log.Infof("CloudCost: ingestor: is already stopping")
+		return
+	}
+
+	msg := "Stopping"
+
+	// If the processes are running (and thus there are channels available for
+	// stopping them) then stop all sub-processes (i.e. build and run)
+	var wg sync.WaitGroup
+
+	if ing.exitBuildCh != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ing.exitBuildCh <- msg
+		}()
+	}
+
+	if ing.exitRunCh != nil {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ing.exitRunCh <- msg
+		}()
+	}
+
+	wg.Wait()
+
+	// Declare that the store is officially no longer running. This allows
+	// Start to be called again, restarting the store from scratch.
+	ing.isRunning.Store(false)
+	ing.isStopping.Store(false)
+}
+
+// Status returns an IngestorStatus that describes the current state of the ingestor
+func (ing *CustomCostIngestor) Status() IngestorStatus {
+	return IngestorStatus{
+		Created:  ing.creationTime,
+		LastRun:  ing.lastRun,
+		NextRun:  ing.lastRun.Add(ing.config.RefreshRate).UTC(),
+		Runs:     ing.runs,
+		Coverage: ing.coverage,
+	}
+}
+
+func (ing *CustomCostIngestor) build(rebuild bool) {
+	// defer errors.HandlePanic()
+
+	// // Profile the full Duration of the build time
+	// buildStart := time.Now()
+
+	// // Build as far back as the configures build Duration
+	// limit := opencost.RoundBack(time.Now().UTC().Add(-ing.config.Duration), ing.config.Resolution)
+
+	// queryWindowStr := timeutil.FormatStoreResolution(ing.config.QueryWindow)
+	// log.Infof("CloudCost[%s]: ingestor: build[%s]: Starting build back to %s in blocks of %s", ing.key, ing.runID, limit.String(), queryWindowStr)
+
+	// // Start with a window of the configured Duration and ending on the given
+	// // start time. Build windows repeating until the window reaches the
+	// // given limit time
+
+	// // Round end times back to nearest Resolution points in the past,
+	// // querying for exactly one interval
+	// e := opencost.RoundBack(time.Now().UTC(), ing.config.Resolution)
+	// s := e.Add(-ing.config.QueryWindow)
+
+	// // Continue until limit is reached
+	// for limit.Before(e) {
+	// 	// If exit instruction is received, log and return
+	// 	select {
+	// 	case <-ing.exitBuildCh:
+	// 		log.Debugf("CloudCost[%s]: ingestor: build[%s]: exiting", ing.key, ing.runID)
+	// 		return
+	// 	default:
+	// 	}
+
+	// 	// Profile the current build step
+	// 	stepStart := time.Now()
+
+	// 	// if rebuild is not specified then check for existing coverage on window
+	// 	if rebuild {
+	// 		ing.BuildWindow(s, e)
+	// 	} else {
+	// 		ing.LoadWindow(s, e)
+	// 	}
+
+	// 	log.Infof("CloudCost[%s]: ingestor: build[%s]:  %s in %v", ing.key, ing.runID, opencost.NewClosedWindow(s, e), time.Since(stepStart))
+
+	// 	// Shift to next QueryWindow
+	// 	s = s.Add(-ing.config.QueryWindow)
+	// 	if s.Before(limit) {
+	// 		s = limit
+	// 	}
+	// 	e = e.Add(-ing.config.QueryWindow)
+	// }
+
+	// log.Infof(fmt.Sprintf("CloudCost[%s]: ingestor: build[%s]: completed in %v", ing.key, ing.runID, time.Since(buildStart)))
+
+	// // In order to be able to Stop, we have to wait on an exit message
+	// // here
+	// <-ing.exitBuildCh
+
+}
+
+func (ing *CustomCostIngestor) Rebuild(domain string) error {
+	return nil
+}
+func (ing *CustomCostIngestor) run() {
+	// defer errors.HandlePanic()
+
+	// ticker := timeutil.NewJobTicker()
+	// defer ticker.Close()
+	// ticker.TickIn(0)
+
+	// for {
+	// 	// If an exit instruction is received, break the run loop
+	// 	select {
+	// 	case <-ing.exitRunCh:
+	// 		log.Debugf("CloudCost[%s]: ingestor: Run[%s] exiting", ing.key, ing.runID)
+	// 		return
+	// 	case <-ticker.Ch:
+	// 		// Wait for next tick
+	// 	}
+
+	// 	// Start from the last covered time, minus the RunWindow
+	// 	start := ing.lastRun
+	// 	start = start.Add(-ing.config.RunWindow)
+
+	// 	// Every Nth (determined by the MonthToDateRunInterval) run should be a month to date run. Where the start is
+	// 	// truncated to the beginning of its current month this can mean that early in a new month we will build all of
+	// 	// last month and the first few days of the current month.
+	// 	if ing.runs%ing.config.MonthToDateRunInterval == 0 {
+	// 		start = time.Date(start.Year(), start.Month(), 1, 0, 0, 0, 0, time.UTC)
+	// 		log.Infof("CloudCost[%s]: ingestor: Run[%s]: running month-to-date update starting at %s", ing.key, ing.runID, start.String())
+	// 	}
+
+	// 	// Round start time back to the nearest Resolution point in the past from the
+	// 	// last update to the QueryWindow
+	// 	s := opencost.RoundBack(start.UTC(), ing.config.Resolution)
+	// 	e := s.Add(ing.config.QueryWindow)
+
+	// 	// Start with a window of the configured Duration and starting on the given
+	// 	// start time. Do the following, repeating until the window reaches the
+	// 	// current time:
+	// 	// 1. Instruct builder to build window
+	// 	// 2. Move window forward one Resolution
+	// 	for time.Now().After(s) {
+	// 		profStart := time.Now()
+	// 		ing.BuildWindow(s, e)
+
+	// 		log.Debugf("CloudCost[%s]: ingestor: Run[%s]: completed %s in %v", ing.key, ing.runID, opencost.NewWindow(&s, &e), time.Since(profStart))
+
+	// 		s = s.Add(ing.config.QueryWindow)
+	// 		e = e.Add(ing.config.QueryWindow)
+	// 		// prevent builds into the future
+	// 		if e.After(time.Now().UTC()) {
+	// 			e = opencost.RoundForward(time.Now().UTC(), ing.config.Resolution)
+	// 		}
+
+	// 	}
+	// 	ing.lastRun = time.Now().UTC()
+
+	// 	limit := opencost.RoundBack(time.Now().UTC(), ing.config.Resolution).Add(-ing.config.Duration)
+	// 	err := ing.repo.Expire(limit)
+	// 	if err != nil {
+	// 		log.Errorf("CloudCost: Ingestor: failed to expire Data: %s", err)
+	// 	}
+
+	// 	ing.coverageLock.Lock()
+	// 	ing.coverage = ing.coverage.ContractStart(limit)
+	// 	ing.coverageLock.Unlock()
+
+	// 	ing.runs++
+
+	// 	ticker.TickIn(ing.config.RefreshRate)
+	// }
+}
+
+func (ing *CustomCostIngestor) expandCoverage(window opencost.Window) {
+	if window.IsOpen() {
+		return
+	}
+	ing.coverageLock.Lock()
+	defer ing.coverageLock.Unlock()
+
+	coverage := ing.coverage.ExpandStart(*window.Start())
+	coverage = coverage.ExpandEnd(*window.End())
+
+	ing.coverage = coverage
+}
+
+// temporary mock data
+const ddData = `
+[
+    {
+        "Metadata": {
+            "api_client_version": "v2"
+        },
+        "Costsource": "observability",
+        "Domain": "datadog",
+        "Version": "v1",
+        "Currency": "USD",
+        "Window": {
+            "start": "2024-02-27T00:00:00Z",
+            "end": "2024-02-27T01:00:00Z"
+        },
+        "Costs": [
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "350+ integrations, alerting, custom metrics \u0026 unlimited user accounts",
+                "ListCost": 90,
+                "ListUnitPrice": 18,
+                "ResourceName": "agent_host_count",
+                "ResourceType": "infra_hosts",
+                "Id": "4bba680574ac970cfba52a5edc5b2d44541319b365fbcc45023b51fbe2572373",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/agent_host_count",
+                "Window": {
+                    "start": "2024-02-27T00:00:00Z",
+                    "end": "2024-02-27T01:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 5,
+                "UsageUnit": "Infra Hosts",
+                "ExtendedAttributes": null
+            },
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "Centralize your monitoring of systems and services (Per Container)",
+                "ListCost": 236,
+                "ListUnitPrice": 1,
+                "ResourceName": "container_count",
+                "ResourceType": "infra_hosts",
+                "Id": "4bba680574ac970cfba52a5edc5b2d44541319b365fbcc45023b51fbe2572373",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/container_count",
+                "Window": {
+                    "start": "2024-02-27T00:00:00Z",
+                    "end": "2024-02-27T01:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 236,
+                "UsageUnit": "Containers",
+                "ExtendedAttributes": null
+            },
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "Centralize your monitoring of systems and services (Per Container)",
+                "ListCost": 219,
+                "ListUnitPrice": 1,
+                "ResourceName": "container_count_excl_agent",
+                "ResourceType": "infra_hosts",
+                "Id": "4bba680574ac970cfba52a5edc5b2d44541319b365fbcc45023b51fbe2572373",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/container_count_excl_agent",
+                "Window": {
+                    "start": "2024-02-27T00:00:00Z",
+                    "end": "2024-02-27T01:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 219,
+                "UsageUnit": "Containers",
+                "ExtendedAttributes": null
+            },
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "350+ integrations, alerting, custom metrics \u0026 unlimited user accounts",
+                "ListCost": 90,
+                "ListUnitPrice": 18,
+                "ResourceName": "host_count",
+                "ResourceType": "infra_hosts",
+                "Id": "4bba680574ac970cfba52a5edc5b2d44541319b365fbcc45023b51fbe2572373",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/host_count",
+                "Window": {
+                    "start": "2024-02-27T00:00:00Z",
+                    "end": "2024-02-27T01:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 5,
+                "UsageUnit": "Infra Hosts",
+                "ExtendedAttributes": null
+            }
+        ],
+        "Errors": null
+    },
+    {
+        "Metadata": {
+            "api_client_version": "v2"
+        },
+        "Costsource": "observability",
+        "Domain": "datadog",
+        "Version": "v1",
+        "Currency": "USD",
+        "Window": {
+            "start": "2024-02-27T01:00:00Z",
+            "end": "2024-02-27T02:00:00Z"
+        },
+        "Costs": [
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "350+ integrations, alerting, custom metrics \u0026 unlimited user accounts",
+                "ListCost": 90,
+                "ListUnitPrice": 18,
+                "ResourceName": "agent_host_count",
+                "ResourceType": "infra_hosts",
+                "Id": "448c8561d845b42adb1d52ebc88b3c44385372e54bf117544442d25887e3c338",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/agent_host_count",
+                "Window": {
+                    "start": "2024-02-27T01:00:00Z",
+                    "end": "2024-02-27T02:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 5,
+                "UsageUnit": "Infra Hosts",
+                "ExtendedAttributes": null
+            },
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "Centralize your monitoring of systems and services (Per Container)",
+                "ListCost": 235,
+                "ListUnitPrice": 1,
+                "ResourceName": "container_count",
+                "ResourceType": "infra_hosts",
+                "Id": "448c8561d845b42adb1d52ebc88b3c44385372e54bf117544442d25887e3c338",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/container_count",
+                "Window": {
+                    "start": "2024-02-27T01:00:00Z",
+                    "end": "2024-02-27T02:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 235,
+                "UsageUnit": "Containers",
+                "ExtendedAttributes": null
+            },
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "Centralize your monitoring of systems and services (Per Container)",
+                "ListCost": 218,
+                "ListUnitPrice": 1,
+                "ResourceName": "container_count_excl_agent",
+                "ResourceType": "infra_hosts",
+                "Id": "448c8561d845b42adb1d52ebc88b3c44385372e54bf117544442d25887e3c338",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/container_count_excl_agent",
+                "Window": {
+                    "start": "2024-02-27T01:00:00Z",
+                    "end": "2024-02-27T02:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 218,
+                "UsageUnit": "Containers",
+                "ExtendedAttributes": null
+            },
+            {
+                "Metadata": null,
+                "Zone": "us",
+                "BilledCost": 0,
+                "AccountName": "Kubecost",
+                "ChargeCategory": "usage",
+                "Description": "350+ integrations, alerting, custom metrics \u0026 unlimited user accounts",
+                "ListCost": 90,
+                "ListUnitPrice": 18,
+                "ResourceName": "host_count",
+                "ResourceType": "infra_hosts",
+                "Id": "448c8561d845b42adb1d52ebc88b3c44385372e54bf117544442d25887e3c338",
+                "ProviderId": "42c0ac62-8d80-11ed-96f3-da7ad0900005/host_count",
+                "Window": {
+                    "start": "2024-02-27T01:00:00Z",
+                    "end": "2024-02-27T02:00:00Z"
+                },
+                "Labels": {},
+                "UsageQty": 5,
+                "UsageUnit": "Infra Hosts",
+                "ExtendedAttributes": null
+            }
+        ],
+        "Errors": null
+    }
+]
+`

pkg/customcost/memoryrepository.go

@@ -0,0 +1,112 @@
+package customcost
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/opencost/opencost/core/pkg/model"
+	"golang.org/x/exp/maps"
+)
+
+// MemoryRepository is an implementation of Repository that uses a map keyed on config key and window start along with a
+// RWMutex to make it threadsafe
+type MemoryRepository struct {
+	rwLock sync.RWMutex
+	data   map[string]map[time.Time][]*model.CustomCostResponse
+}
+
+func NewMemoryRepository() *MemoryRepository {
+	return &MemoryRepository{
+		data: make(map[string]map[time.Time][]*model.CustomCostResponse),
+	}
+}
+
+func (m *MemoryRepository) Has(startTime time.Time, domain string) (bool, error) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
+	domainData, ok := m.data[domain]
+	if !ok {
+		return false, nil
+	}
+
+	_, ook := domainData[startTime.UTC()]
+	return ook, nil
+}
+
+func (m *MemoryRepository) Get(startTime time.Time, domain string) ([]*model.CustomCostResponse, error) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
+	domainData, ok := m.data[domain]
+	if !ok {
+		return nil, nil
+	}
+
+	ccr, ook := domainData[startTime.UTC()]
+	if !ook {
+		return nil, nil
+	}
+	clones := []*model.CustomCostResponse{}
+
+	for _, cc := range ccr {
+		clone := cc.Clone()
+		clones = append(clones, &clone)
+	}
+
+	return clones, nil
+}
+
+func (m *MemoryRepository) Keys() ([]string, error) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
+	keys := maps.Keys(m.data)
+	return keys, nil
+}
+
+func (m *MemoryRepository) Put(ccr []*model.CustomCostResponse) error {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
+	if ccr == nil {
+		return fmt.Errorf("MemoryRepository: Put: cannot save nil")
+	}
+
+	for _, cc := range ccr {
+		if cc.Window.IsOpen() {
+			return fmt.Errorf("MemoryRepository: Put: custom cost response has invalid window %s", cc.Window.String())
+		}
+
+		if cc.GetDomain() == "" {
+			return fmt.Errorf("MemoryRepository: Put: custom cost response does not have a domain value")
+		}
+
+		if _, ok := m.data[cc.GetDomain()]; !ok {
+			m.data[cc.GetDomain()] = make(map[time.Time][]*model.CustomCostResponse)
+		}
+
+		m.data[cc.GetDomain()][cc.Window.Start().UTC()] = append(m.data[cc.GetDomain()][cc.Window.Start().UTC()], cc)
+	}
+	return nil
+}
+
+// Expire deletes all items in the map with a start time before the given limit
+func (m *MemoryRepository) Expire(limit time.Time) error {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
+	for key, integration := range m.data {
+		for startTime := range integration {
+			if startTime.Before(limit) {
+				delete(integration, startTime)
+			}
+		}
+		// remove integration if it is now empty
+		if len(integration) == 0 {
+			delete(m.data, key)
+		}
+	}
+	return nil
+}

pkg/customcost/pipelineservice.go

@@ -0,0 +1,139 @@
+package customcost
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/hashicorp/go-plugin"
+	"github.com/julienschmidt/httprouter"
+	"github.com/opencost/opencost/core/pkg/log"
+	proto "github.com/opencost/opencost/core/pkg/protocol"
+	"github.com/opencost/opencost/pkg/env"
+)
+
+var protocol = proto.HTTP()
+
+// PipelineService exposes CloudCost pipeline controls and diagnostics endpoints
+type PipelineService struct {
+	hourlyIngestor, dailyIngestor *CustomCostIngestor
+	hourlyStore, dailyStore       Repository
+}
+
+func getRegisteredPlugins(configDir string, execDir string) map[string]*plugin.Client {
+	plugins := map[string]*plugin.Client{}
+	return plugins
+}
+
+// NewPipelineService is a constructor for a PipelineService
+func NewPipelineService(hourlyrepo, dailyrepo Repository, ingConf CustomCostIngestorConfig) (*PipelineService, error) {
+
+	registeredPlugins := getRegisteredPlugins(ingConf.PluginConfigDir, ingConf.PluginExecutableDir)
+
+	hourlyIngestor, err := NewCustomCostIngestor(&ingConf, hourlyrepo, registeredPlugins)
+	if err != nil {
+		return nil, err
+	}
+
+	hourlyIngestor.Start(false)
+
+	dailyIngestor, err := NewCustomCostIngestor(&ingConf, dailyrepo, registeredPlugins)
+	if err != nil {
+		return nil, err
+	}
+
+	dailyIngestor.Start(false)
+	return &PipelineService{
+		hourlyIngestor: hourlyIngestor,
+		hourlyStore:    hourlyrepo,
+		dailyStore:     dailyrepo,
+		dailyIngestor:  dailyIngestor,
+	}, nil
+}
+
+// Status gives a combined view of the state of configs and the ingestior status
+func (dp *PipelineService) Status() Status {
+
+	// Pull config status from the config controller
+	ingstatus := dp.hourlyIngestor.Status()
+	dur, err := time.ParseDuration(env.GetCustomCostRefreshRateHours())
+	if err != nil {
+		log.Errorf("error parsing duration %s: %v", env.GetCustomCostRefreshRateHours(), err)
+		return Status{}
+	}
+	refreshRate := time.Hour * dur
+
+	// These are the statuses
+	return Status{
+		Coverage:    ingstatus.Coverage.String(),
+		RefreshRate: refreshRate.String(),
+	}
+
+}
+
+// GetCloudCostRebuildHandler creates a handler from a http request which initiates a rebuild of cloud cost pipeline, if an
+// integrationKey is provided then it only rebuilds the specified billing integration
+func (s *PipelineService) GetCloudCostRebuildHandler() func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+	// If pipeline Service is nil, always return 501
+	if s == nil {
+		return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+			http.Error(w, "Custom Cost Pipeline Service is nil", http.StatusNotImplemented)
+		}
+	}
+	if s.dailyIngestor == nil || s.hourlyIngestor == nil {
+		return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+			http.Error(w, "Custom Cost Pipeline Service Ingestion Manager is nil", http.StatusNotImplemented)
+		}
+	}
+	// Return valid handler func
+	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+		w.Header().Set("Content-Type", "application/json")
+
+		commit := r.URL.Query().Get("commit") == "true" || r.URL.Query().Get("commit") == "1"
+
+		if !commit {
+			protocol.WriteData(w, "Pass parameter 'commit=true' to confirm Cloud Cost rebuild")
+			return
+		}
+
+		domain := r.URL.Query().Get("domain")
+
+		err := s.hourlyIngestor.Rebuild(domain)
+		if err != nil {
+			log.Errorf("error rebuilding hourly ingestor")
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		err = s.dailyIngestor.Rebuild(domain)
+		if err != nil {
+			log.Errorf("error rebuilding daily ingestor")
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		protocol.WriteData(w, fmt.Sprintf("Rebuilding Custom Cost For Domain %s", domain))
+		return
+
+	}
+}
+
+// GetCloudCostStatusHandler creates a handler from a http request which returns a list of the billing integration status
+func (s *PipelineService) GetCustomCostStatusHandler() func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+	// If Reporting Service is nil, always return 501
+	if s == nil {
+		return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+			http.Error(w, "Reporting Service is nil", http.StatusNotImplemented)
+		}
+	}
+	if s.hourlyIngestor == nil || s.dailyIngestor == nil {
+		return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+			http.Error(w, "Custom Cost Pipeline Service Ingestor is nil", http.StatusNotImplemented)
+		}
+	}
+
+	// Return valid handler func
+	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+		w.Header().Set("Content-Type", "application/json")
+
+		protocol.WriteData(w, s.Status())
+	}
+}

pkg/customcost/repository.go

@@ -0,0 +1,16 @@
+package customcost
+
+import (
+	"time"
+
+	"github.com/opencost/opencost/core/pkg/model"
+)
+
+// Repository is an interface for storing and retrieving CloudCost data
+type Repository interface {
+	Has(time.Time, string) (bool, error)
+	Get(time.Time, string) ([]*model.CustomCostResponse, error)
+	Keys() ([]string, error)
+	Put([]*model.CustomCostResponse) error
+	Expire(time.Time) error
+}

pkg/customcost/status.go

@@ -0,0 +1,24 @@
+package customcost
+
+import (
+	"time"
+
+	cloudconfig "github.com/opencost/opencost/pkg/cloud"
+)
+
+// Status gives the details and metadata of a CloudCost integration
+type Status struct {
+	Key              string             `json:"key"`
+	Source           string             `json:"source"`
+	Provider         string             `json:"provider"`
+	Active           bool               `json:"active"`
+	Valid            bool               `json:"valid"`
+	LastRun          time.Time          `json:"lastRun"`
+	NextRun          time.Time          `json:"nextRun"`
+	RefreshRate      string             `json:"RefreshRate"`
+	Created          time.Time          `json:"created"`
+	Runs             int                `json:"runs"`
+	Coverage         string             `json:"coverage"`
+	ConnectionStatus string             `json:"connectionStatus"`
+	Config           cloudconfig.Config `json:"config"`
+}

pkg/env/costmodelenv.go

@@ -114,7 +114,8 @@ const (
 	ExportCSVLabelsAll  = "EXPORT_CSV_LABELS_ALL"
 	ExportCSVMaxDays    = "EXPORT_CSV_MAX_DAYS"
 
-	DataRetentionDailyResolutionDaysEnvVar = "DATA_RETENTION_DAILY_RESOLUTION_DAYS"
+	DataRetentionDailyResolutionDaysEnvVar   = "DATA_RETENTION_DAILY_RESOLUTION_DAYS"
+	DataRetentionHourlyResolutionHoursEnvVar = "DATA_RETENTION_HOURLY_RESOLUTION_HOURS"
 
 	// We assume that Kubernetes is enabled if there is a KUBERNETES_PORT environment variable present
 	KubernetesEnabledEnvVar         = "KUBERNETES_PORT"
@@ -125,6 +126,13 @@ const (
 	CloudCostQueryWindowDaysEnvVar  = "CLOUD_COST_QUERY_WINDOW_DAYS"
 	CloudCostRunWindowDaysEnvVar    = "CLOUD_COST_RUN_WINDOW_DAYS"
 
+	CustomCostEnabledEnvVar          = "CUSTOM_COST_ENABLED"
+	CustomCostQueryWindowDaysEnvVar  = "CUSTOM_COST_QUERY_WINDOW_DAYS"
+	CustomCostRefreshRateHoursEnvVar = "CUSTOM_COST_REFRESH_RATE_HOURS"
+
+	PluginConfigDirEnvVar     = "PLUGIN_CONFIG_DIR"
+	PluginExecutableDirEnvVar = "PLUGIN_EXECUTABLE_DIR"
+
 	OCIPricingURL = "OCI_PRICING_URL"
 )
 
@@ -637,6 +645,10 @@ func GetDataRetentionDailyResolutionDays() int64 {
 	return env.GetInt64(DataRetentionDailyResolutionDaysEnvVar, 15)
 }
 
+func GetDataRetentionHourlyResolutionHours() int64 {
+	return env.GetInt64(DataRetentionHourlyResolutionHoursEnvVar, 49)
+}
+
 func IsKubernetesEnabled() bool {
 	return env.Get(KubernetesEnabledEnvVar, "") != ""
 }
@@ -645,6 +657,10 @@ func IsCloudCostEnabled() bool {
 	return env.GetBool(CloudCostEnabledEnvVar, false)
 }
 
+func IsCustomCostEnabled() bool {
+	return env.GetBool(CustomCostEnabledEnvVar, false)
+}
+
 func GetCloudCostConfigPath() string {
 	return env.Get(CloudCostConfigPath, "cloud-integration.json")
 }
@@ -661,6 +677,10 @@ func GetCloudCostQueryWindowDays() int64 {
 	return env.GetInt64(CloudCostQueryWindowDaysEnvVar, 7)
 }
 
+func GetCustomCostQueryWindowDays() int64 {
+	return env.GetInt64(CustomCostQueryWindowDaysEnvVar, 7)
+}
+
 func GetCloudCostRunWindowDays() int64 {
 	return env.GetInt64(CloudCostRunWindowDaysEnvVar, 3)
 }
@@ -668,3 +688,15 @@ func GetCloudCostRunWindowDays() int64 {
 func GetOCIPricingURL() string {
 	return env.Get(OCIPricingURL, "https://apexapps.oracle.com/pls/apex/cetools/api/v1/products")
 }
+
+func GetPluginConfigDir() string {
+	return env.Get(PluginConfigDirEnvVar, "/opt/opencost/plugin/config")
+}
+
+func GetPluginExecutableDir() string {
+	return env.Get(PluginExecutableDirEnvVar, "/opt/opencost/plugin/bin")
+}
+
+func GetCustomCostRefreshRateHours() string {
+	return env.Get(CustomCostRefreshRateHoursEnvVar, "12h")
+}