Updates for exporter

Signed-off-by: Sean Holcomb <seanholcomb@gmail.com>

core/pkg/env/core.go

@@ -20,6 +20,9 @@ const (
 	Resolution1dRetentionEnvVar  = "RESOLUTION_1D_RETENTION"  // int: number of days
 	Resolution1hRetentionEnvVar  = "RESOLUTION_1H_RETENTION"  // int: number of hours
 	Resolution10mRetentionEnvVar = "RESOLUTION_10M_RETENTION" // int: number of 10m segments
+
+	ExportLegacyDataModelEnvVar = "EXPORT_LEGACY_DATA_MODEL"
+	ExportKubeModelEnvVar       = "EXPORT_KUBEMODEL"
 )
 
 // GetAPIPort returns the environment variable value for APIPortEnvVar which
@@ -61,3 +64,11 @@ func IsPProfEnabled() bool {
 func GetInstallNamespace(def string) string {
 	return Get(InstallNamespaceEnvVar, def)
 }
+
+func GetExportLegacyDataModel() bool {
+	return GetBool(ExportLegacyDataModelEnvVar, true)
+}
+
+func GetExportKubeModel() bool {
+	return GetBool(ExportKubeModelEnvVar, true)
+}

core/pkg/opencost/exporter/controllers.go

@@ -42,6 +42,17 @@ type PipelinesExportConfig struct {
 	Compression                       ExportCompressionLevel
 }
 
+func (c PipelinesExportConfig) toControllerConfig(res time.Duration) ComputeExporterConfig {
+	return ComputeExporterConfig{
+		AppName:     c.AppName,
+		ClusterUID:  c.ClusterUID,
+		ClusterName: c.ClusterName,
+		Resolution:  res,
+		Streaming:   c.Streaming,
+		Compression: c.Compression,
+	}
+}
+
 // defaultPipelineExportResolutions returns the default export configuration for the pipeline
 // which is set to export hourly and daily.
 func defaultPipelineExportResolutions() []time.Duration {
@@ -53,18 +64,24 @@ func defaultPipelineExportResolutions() []time.Duration {
 
 // NewPipelinesExportConfig returns the default export configuration for all pipelines
 // which is set to export hourly and daily for allocations, assets, and network insights.
-func NewPipelinesExportConfig(appName, clusterUID, clusterName string) PipelinesExportConfig {
-	return PipelinesExportConfig{
-		AppName:                           appName,
-		ClusterUID:                        clusterUID,
-		ClusterName:                       clusterName,
-		AllocationPiplineResolutions:      defaultPipelineExportResolutions(),
-		AssetPipelineResolutons:           defaultPipelineExportResolutions(),
-		NetworkInsightPipelineResolutions: defaultPipelineExportResolutions(),
-		KubeModelPipelineResolutions:      defaultPipelineExportResolutions(),
-		Streaming:                         false,
-		Compression:                       ExportCompressionLevelNone,
+func NewPipelinesExportConfig(appName, clusterUID, clusterName string, exportLegacy, exportKubeModel bool) PipelinesExportConfig {
+	config := PipelinesExportConfig{
+		AppName:     appName,
+		ClusterUID:  clusterUID,
+		ClusterName: clusterName,
+		Compression: ExportCompressionLevelNone,
 	}
+	if exportLegacy {
+		config.AllocationPiplineResolutions = defaultPipelineExportResolutions()
+		config.AssetPipelineResolutons = defaultPipelineExportResolutions()
+		config.NetworkInsightPipelineResolutions = defaultPipelineExportResolutions()
+
+	}
+	if exportKubeModel {
+		config.KubeModelPipelineResolutions = defaultPipelineExportResolutions()
+	}
+
+	return config
 }
 
 // PipelineExportControllers is a facade that contains the export controllers for allocations, assets, and network insights.
@@ -96,14 +113,9 @@ func NewPipelineExportControllers(store storage.Storage, cm ComputePipelineSourc
 			continue
 		}
 
-		// Use ClusterName for "clusterId" here to maintain legacy pattern
-		var allocController *export.ComputeExportController[opencost.AllocationSet]
-		var err error
-		if config.Streaming {
-			allocController, err = NewStreamingComputePipelineExportController(config.ClusterName, store, allocSource, res, config.Compression)
-		} else {
-			allocController, err = NewComputePipelineExportController(config.ClusterName, store, allocSource, res)
-		}
+		controllerConfig := config.toControllerConfig(res)
+		allocController, err := NewComputePipelineExportController(controllerConfig, store, allocSource)
+
 		if err != nil {
 			log.Errorf("Failed to create allocation export controller for resolution: %s - %v", timeutil.DurationString(res), err)
 			continue
@@ -122,14 +134,9 @@ func NewPipelineExportControllers(store storage.Storage, cm ComputePipelineSourc
 			continue
 		}
 
-		// Use ClusterName for "clusterId" here to maintain legacy pattern
-		var assetController *export.ComputeExportController[opencost.AssetSet]
-		var err error
-		if config.Streaming {
-			assetController, err = NewStreamingComputePipelineExportController(config.ClusterName, store, assetSource, res, config.Compression)
-		} else {
-			assetController, err = NewComputePipelineExportController(config.ClusterName, store, assetSource, res)
-		}
+		controllerConfig := config.toControllerConfig(res)
+		assetController, err := NewComputePipelineExportController(controllerConfig, store, assetSource)
+
 		if err != nil {
 			log.Errorf("Failed to create asset export controller for resolution: %s - %v", timeutil.DurationString(res), err)
 			continue
@@ -148,14 +155,9 @@ func NewPipelineExportControllers(store storage.Storage, cm ComputePipelineSourc
 			continue
 		}
 
-		// Use ClusterName for "clusterId" here to maintain legacy pattern
-		var networkInsightController *export.ComputeExportController[opencost.NetworkInsightSet]
-		var err error
-		if config.Streaming {
-			networkInsightController, err = NewStreamingComputePipelineExportController(config.ClusterName, store, networkInsightSource, res, config.Compression)
-		} else {
-			networkInsightController, err = NewComputePipelineExportController(config.ClusterName, store, networkInsightSource, res)
-		}
+		controllerConfig := config.toControllerConfig(res)
+		networkInsightController, err := NewComputePipelineExportController(controllerConfig, store, networkInsightSource)
+
 		if err != nil {
 			log.Errorf("Failed to create network insight export controller for resolution: %s - %v", timeutil.DurationString(res), err)
 			continue
@@ -174,13 +176,8 @@ func NewPipelineExportControllers(store storage.Storage, cm ComputePipelineSourc
 			continue
 		}
 
-		var kubeModelController *export.ComputeExportController[kubemodel.KubeModelSet]
-		var err error
-		if config.Streaming {
-			kubeModelController, err = NewStreamingKubeModelComputePipelineExportController(config.AppName, config.ClusterUID, store, kubeModelSource, res, config.Compression)
-		} else {
-			kubeModelController, err = NewKubeModelComputePipelineExportController(config.AppName, config.ClusterUID, store, kubeModelSource, res)
-		}
+		controllerConfig := config.toControllerConfig(res)
+		kubeModelController, err := NewComputePipelineExportController(controllerConfig, store, kubeModelSource)
 
 		if err != nil {
 			log.Errorf("Failed to create KubeModel export controller for resolution: %s - %v", timeutil.DurationString(res), err)

core/pkg/opencost/exporter/exporter_test.go

@@ -172,7 +172,7 @@ func TestExporters(t *testing.T) {
 			t.Fatalf("failed to create path formatter: %v", err)
 		}
 
-		allocExporter, err := NewComputePipelineExporter[opencost.AllocationSet](TestClusterName, TestResolution, memStore)
+		allocExporter, err := NewComputePipelineExporter[opencost.AllocationSet](ComputeExporterConfig{AppName: TestAppName, ClusterUID: TestClusterID, ClusterName: TestClusterName, Resolution: TestResolution}, memStore)
 		if err != nil {
 			t.Fatalf("failed to create allocation exporter: %v", err)
 		}
@@ -201,7 +201,7 @@ func TestExporters(t *testing.T) {
 			t.Fatalf("failed to create path formatter: %v", err)
 		}
 
-		assetExporter, err := NewComputePipelineExporter[opencost.AssetSet](TestClusterName, TestResolution, memStore)
+		assetExporter, err := NewComputePipelineExporter[opencost.AssetSet](ComputeExporterConfig{AppName: TestAppName, ClusterUID: TestClusterID, ClusterName: TestClusterName, Resolution: TestResolution}, memStore)
 		if err != nil {
 			t.Fatalf("failed to create allocation exporter: %v", err)
 		}
@@ -230,7 +230,7 @@ func TestExporters(t *testing.T) {
 			t.Fatalf("failed to create path formatter: %v", err)
 		}
 
-		netInsightExporter, err := NewComputePipelineExporter[opencost.NetworkInsightSet](TestClusterName, TestResolution, memStore)
+		netInsightExporter, err := NewComputePipelineExporter[opencost.NetworkInsightSet](ComputeExporterConfig{AppName: TestAppName, ClusterUID: TestClusterID, ClusterName: TestClusterName, Resolution: TestResolution}, memStore)
 		if err != nil {
 			t.Fatalf("failed to create net insights exporter: %v", err)
 		}
@@ -260,7 +260,7 @@ func TestExporters(t *testing.T) {
 			t.Fatalf("failed to create path formatter: %v", err)
 		}
 
-		kubeModelExporter, err := NewKubeModelComputePipelineExporter[kubemodel.KubeModelSet](TestAppName, TestClusterID, TestResolution, memStore)
+		kubeModelExporter, err := NewComputePipelineExporter[kubemodel.KubeModelSet](ComputeExporterConfig{AppName: TestAppName, ClusterUID: TestClusterID, Resolution: TestResolution}, memStore)
 		if err != nil {
 			t.Fatalf("failed to create KubeModel exporter: %v", err)
 		}
@@ -278,20 +278,20 @@ func TestExporters(t *testing.T) {
 			t.Fatalf("failed to export KubeModel data: %v", err)
 		}
 
-		validateFileCreation[kubemodel.KubeModelSet](t, memStore, p, exporter.BingenExt, start, end)
+		validateFileCreation[kubemodel.KubeModelSet](t, memStore, p, "", start, end)
 	})
 
 	t.Run("unknown exporter", func(t *testing.T) {
 		memStore := storage.NewMemoryStorage()
 
 		// Invalid pipeline
-		_, err := NewComputePipelineExporter[UnknownSet](TestClusterName, TestResolution, memStore)
+		_, err := NewComputePipelineExporter[UnknownSet](ComputeExporterConfig{ClusterName: TestClusterName, Resolution: TestResolution}, memStore)
 		if err == nil {
 			t.Fatalf("expected error creating unknown pipeline exporter, got nil")
 		}
 
 		// Invalid cluster id
-		_, err = NewComputePipelineExporter[opencost.AllocationSet]("", TestResolution, memStore)
+		_, err = NewComputePipelineExporter[opencost.AllocationSet](ComputeExporterConfig{ClusterName: "", Resolution: TestResolution}, memStore)
 		if err == nil {
 			t.Fatalf("expected error creating allocation pipeline exporter with empty cluster id, got nil")
 		}
@@ -381,7 +381,7 @@ func TestPipelineExportControllers(t *testing.T) {
 		pipelineComputeSource := NewMockPipelineComputeSource()
 		memStore := storage.NewMemoryStorage()
 
-		exportControllers := NewPipelineExportControllers(memStore, pipelineComputeSource, NewPipelinesExportConfig(TestAppName, TestClusterID, TestClusterName))
+		exportControllers := NewPipelineExportControllers(memStore, pipelineComputeSource, NewPipelinesExportConfig(TestAppName, TestClusterID, TestClusterName, true, false))
 
 		if len(exportControllers.AllocationExportController.Resolutions()) != 2 {
 			t.Fatalf("expected 2 allocation resolutions, got %d", len(exportControllers.AllocationExportController.Resolutions()))
@@ -399,7 +399,7 @@ func TestPipelineExportControllers(t *testing.T) {
 		pipelineComputeSource := NewMockPipelineComputeSourceWith(48 * time.Hour)
 		memStore := storage.NewMemoryStorage()
 
-		exportControllers := NewPipelineExportControllers(memStore, pipelineComputeSource, NewPipelinesExportConfig(TestAppName, TestClusterID, TestClusterName))
+		exportControllers := NewPipelineExportControllers(memStore, pipelineComputeSource, NewPipelinesExportConfig(TestAppName, TestClusterID, TestClusterName, true, true))
 
 		if len(exportControllers.AllocationExportController.Resolutions()) != 0 {
 			t.Fatalf("expected 0 allocation resolutions, got %d", len(exportControllers.AllocationExportController.Resolutions()))
@@ -416,7 +416,7 @@ func TestPipelineExportControllers(t *testing.T) {
 		pipelineComputeSource := NewMockPipelineComputeSource()
 		memStore := storage.NewMemoryStorage()
 
-		exportControllers := NewPipelineExportControllers(memStore, pipelineComputeSource, NewPipelinesExportConfig("", "", ""))
+		exportControllers := NewPipelineExportControllers(memStore, pipelineComputeSource, NewPipelinesExportConfig("", "", "", false, false))
 
 		if len(exportControllers.AllocationExportController.Resolutions()) != 0 {
 			t.Fatalf("expected 0 allocation resolutions, got %d", len(exportControllers.AllocationExportController.Resolutions()))

core/pkg/opencost/exporter/exporters.go

@@ -8,6 +8,7 @@ import (
 	export "github.com/opencost/opencost/core/pkg/exporter"
 	"github.com/opencost/opencost/core/pkg/exporter/pathing"
 	"github.com/opencost/opencost/core/pkg/exporter/validator"
+	"github.com/opencost/opencost/core/pkg/opencost"
 	"github.com/opencost/opencost/core/pkg/pipelines"
 	"github.com/opencost/opencost/core/pkg/storage"
 	"github.com/opencost/opencost/core/pkg/util/timeutil"
@@ -36,60 +37,34 @@ const (
 	ExportCompressionLevelDefault         ExportCompressionLevel = gzip.DefaultCompression
 )
 
-// NewComputePipelineExporter creates a new `ComputeExporter[T]` instance which is used to export computed data
-// by window for a specific pipeline.
-func NewComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	clusterName string,
-	resolution time.Duration,
-	store storage.Storage,
-) (export.ComputeExporter[T], error) {
-	pipelineName := pipelines.NameFor[T]()
-	if pipelineName == "" {
-		return nil, fmt.Errorf("failed to extract pipeline name for type: %s", typeutil.TypeOf[T]())
-	}
-
-	pathing, err := pathing.NewDefaultStoragePathFormatter(clusterName, pipelineName, &resolution)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create path formatter: %w", err)
-	}
-
-	var encoder export.Encoder[T]
-	encoder = export.NewBingenEncoder[T, U]()
-
-	return export.NewComputeStorageExporter(
-		pathing,
-		encoder,
-		store,
-		validator.NewSetValidator[T, S](resolution),
-		false,
-	), nil
+type ComputeExporterConfig struct {
+	AppName     string
+	ClusterUID  string
+	ClusterName string
+	Resolution  time.Duration
+	Streaming   bool
+	Compression ExportCompressionLevel
 }
 
-// NewStreamingComputePipelineExporter creates a new `ComputeExporter[T]` instance which is used to export computed data
+// NewComputePipelineExporter creates a new `ComputeExporter[T]` instance which is used to export computed data
 // by window for a specific pipeline.
-func NewStreamingComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	clusterId string,
-	resolution time.Duration,
+func NewComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
+	config ComputeExporterConfig,
 	store storage.Storage,
-	compressionLevel ExportCompressionLevel,
 ) (export.ComputeExporter[T], error) {
-	pipelineName := pipelines.NameFor[T]()
-	if pipelineName == "" {
-		return nil, fmt.Errorf("failed to extract pipeline name for type: %s", typeutil.TypeOf[T]())
-	}
 
-	pathing, err := pathing.NewDefaultStoragePathFormatter(clusterId, pipelineName, &resolution)
+	pathing, err := GetExporterPathing[T, U, S](config)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create path formatter: %w", err)
 	}
 
-	if !compressionLevel.IsValid() {
-		return nil, fmt.Errorf("invalid compression level passed: %d is not a valid compression level", int(compressionLevel))
+	if !config.Compression.IsValid() {
+		return nil, fmt.Errorf("invalid compression level passed: %d is not a valid compression level", int(config.Compression))
 	}
 
 	var encoder export.Encoder[T]
-	if compressionLevel != ExportCompressionLevelNone {
-		encoder = export.NewGZipEncoderWithLevel(export.NewBingenEncoder[T, U](), int(compressionLevel))
+	if config.Streaming && config.Compression != ExportCompressionLevelNone {
+		encoder = export.NewGZipEncoderWithLevel(export.NewBingenEncoder[T, U](), int(config.Compression))
 	} else {
 		encoder = export.NewBingenEncoder[T, U]()
 	}
@@ -98,144 +73,48 @@ func NewStreamingComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T],
 		pathing,
 		encoder,
 		store,
-		validator.NewSetValidator[T, S](resolution),
-		true,
+		validator.NewSetValidator[T, S](config.Resolution),
+		config.Streaming,
 	), nil
 }
 
 // NewComputePipelineExportController creates a new `ComputeExportController[T]` instance which is used to export computed data
 // using the provided source, storage, resolution, and source resolution.
 func NewComputePipelineExportController[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	clusterName string,
-	store storage.Storage,
-	source export.ComputeSource[T],
-	resolution time.Duration,
-) (*export.ComputeExportController[T], error) {
-	exporter, err := NewComputePipelineExporter[T, U, S](clusterName, resolution, store)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create compute exporter: %w", err)
-	}
-
-	return export.NewComputeExportController(source, exporter, resolution), nil
-}
-
-// NewStreamingComputePipelineExportController creates a new `ComputeExportController[T]` instance which is used to stream/export the
-// computed data using the provided source, storage, resolution, and source resolution.
-func NewStreamingComputePipelineExportController[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	clusterId string,
+	config ComputeExporterConfig,
 	store storage.Storage,
 	source export.ComputeSource[T],
-	resolution time.Duration,
-	compressionLevel ExportCompressionLevel,
 ) (*export.ComputeExportController[T], error) {
-	exporter, err := NewStreamingComputePipelineExporter[T, U, S](clusterId, resolution, store, compressionLevel)
+	exporter, err := NewComputePipelineExporter[T, U, S](config, store)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create compute exporter: %w", err)
 	}
 
-	return export.NewComputeExportController(source, exporter, resolution), nil
-}
-
-// NewKubeModelComputePipelineExporter creates a new `ComputeExporter[T]` instance which is used to export computed data
-// by window for a specific pipeline.
-func NewKubeModelComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	appName string,
-	clusterId string,
-	resolution time.Duration,
-	store storage.Storage,
-) (export.ComputeExporter[T], error) {
-	pipelineName := pipelines.NameFor[T]()
-	if pipelineName == "" {
-		return nil, fmt.Errorf("failed to extract pipeline name for type: %s", typeutil.TypeOf[T]())
-	}
-	res := timeutil.FormatStoreResolution(resolution)
-	pathing, err := pathing.NewKubeModelStoragePathFormatter(appName, clusterId, res)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create path formatter: %w", err)
-	}
-
-	encoder := export.NewBingenFileEncoder[T, U]()
-
-	return export.NewComputeStorageExporter(
-		pathing,
-		encoder,
-		store,
-		validator.NewSetValidator[T, S](resolution),
-		false,
-	), nil
+	return export.NewComputeExportController(source, exporter, config.Resolution), nil
 }
 
-// NewStreamingComputePipelineExporter creates a new `ComputeExporter[T]` instance which is used to export computed data
-// by window for a specific pipeline.
-func NewStreamingKubeModelComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	appName string,
-	clusterId string,
-	resolution time.Duration,
-	store storage.Storage,
-	compressionLevel ExportCompressionLevel,
-) (export.ComputeExporter[T], error) {
+func GetExporterPathing[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
+	config ComputeExporterConfig,
+) (pathing.StoragePathFormatter[opencost.Window], error) {
 	pipelineName := pipelines.NameFor[T]()
 	if pipelineName == "" {
 		return nil, fmt.Errorf("failed to extract pipeline name for type: %s", typeutil.TypeOf[T]())
 	}
+	res := timeutil.FormatStoreResolution(config.Resolution)
 
-	res := timeutil.FormatStoreResolution(resolution)
-	pathing, err := pathing.NewKubeModelStoragePathFormatter(appName, clusterId, res)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create path formatter: %w", err)
-	}
-
-	if !compressionLevel.IsValid() {
-		return nil, fmt.Errorf("invalid compression level passed: %d is not a valid compression level", int(compressionLevel))
-	}
-
-	var encoder export.Encoder[T]
-	if compressionLevel != ExportCompressionLevelNone {
-		encoder = export.NewGZipEncoderWithLevel(export.NewBingenEncoder[T, U](), int(compressionLevel))
-	} else {
-		encoder = export.NewBingenEncoder[T, U]()
-	}
-
-	return export.NewComputeStorageExporter(
-		pathing,
-		encoder,
-		store,
-		validator.NewSetValidator[T, S](resolution),
-		true,
-	), nil
-}
+	var pathFormatter pathing.StoragePathFormatter[opencost.Window]
+	var err error
 
-// NewKubeModelComputePipelineExportController creates a new `ComputeExportController[T]` instance which is used to export computed data
-// using the provided source, storage, resolution, and source resolution.
-func NewKubeModelComputePipelineExportController[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	appName string,
-	clusterId string,
-	store storage.Storage,
-	source export.ComputeSource[T],
-	resolution time.Duration,
-) (*export.ComputeExportController[T], error) {
-	exporter, err := NewKubeModelComputePipelineExporter[T, U, S](appName, clusterId, resolution, store)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create compute exporter: %w", err)
+	switch pipelineName {
+	case pipelines.KubeModelPipelineName:
+		pathFormatter, err = pathing.NewKubeModelStoragePathFormatter(config.AppName, config.ClusterUID, res)
+	default:
+		// Use ClusterName for "clusterId" here to maintain legacy pattern
+		pathFormatter, err = pathing.NewDefaultStoragePathFormatter(config.ClusterName, pipelineName, &config.Resolution)
 	}
 
-	return export.NewComputeExportController(source, exporter, resolution), nil
-}
-
-// NewStreamingComputePipelineExportController creates a new `ComputeExportController[T]` instance which is used to stream/export the
-// computed data using the provided source, storage, resolution, and source resolution.
-func NewStreamingKubeModelComputePipelineExportController[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
-	appName string,
-	clusterId string,
-	store storage.Storage,
-	source export.ComputeSource[T],
-	resolution time.Duration,
-	compressionLevel ExportCompressionLevel,
-) (*export.ComputeExportController[T], error) {
-	exporter, err := NewStreamingKubeModelComputePipelineExporter[T, U, S](appName, clusterId, resolution, store, compressionLevel)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create compute exporter: %w", err)
+		return nil, fmt.Errorf("failed to create path formatter[%s]: %w", pipelineName, err)
 	}
-
-	return export.NewComputeExportController(source, exporter, resolution), nil
+	return pathFormatter, nil
 }

pkg/kubemodel/pipeline.go

@@ -4,15 +4,15 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/opencost/opencost/core/pkg/env"
 	ocexporter "github.com/opencost/opencost/core/pkg/opencost/exporter"
 	"github.com/opencost/opencost/core/pkg/storage"
 	"github.com/opencost/opencost/core/pkg/util/timeutil"
 )
 
 var (
-	exportInterval     = 5 * time.Minute
-	janitorInterval    = timeutil.Day
-	defaultResolutions = []time.Duration{time.Hour, timeutil.Day}
+	exportInterval  = 5 * time.Minute
+	janitorInterval = timeutil.Day
 )
 
 // Pipeline manages the KubeModel export controller group and the retention janitor.
@@ -30,11 +30,7 @@ func NewPipeline(appName, clusterUID string, store storage.Storage, cm ocexporte
 		return nil, fmt.Errorf("NewPipeline: clusterUID cannot be empty")
 	}
 
-	config := ocexporter.PipelinesExportConfig{
-		AppName:                      appName,
-		ClusterUID:                   clusterUID,
-		KubeModelPipelineResolutions: defaultResolutions,
-	}
+	config := ocexporter.NewPipelinesExportConfig(appName, clusterUID, "", false, env.GetExportKubeModel())
 
 	controllers := ocexporter.NewPipelineExportControllers(store, cm, config)