exporters.go 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. package exporter
  2. import (
  3. "fmt"
  4. "time"
  5. export "github.com/opencost/opencost/core/pkg/exporter"
  6. "github.com/opencost/opencost/core/pkg/exporter/pathing"
  7. "github.com/opencost/opencost/core/pkg/exporter/validator"
  8. "github.com/opencost/opencost/core/pkg/pipelines"
  9. "github.com/opencost/opencost/core/pkg/storage"
  10. "github.com/opencost/opencost/core/pkg/util/typeutil"
  11. )
  12. // NewComputePipelineExporter creates a new `ComputeExporter[T]` instance which is used to export computed data
  13. // by window for a specific pipeline.
  14. func NewComputePipelineExporter[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
  15. clusterId string,
  16. resolution time.Duration,
  17. store storage.Storage,
  18. ) (export.ComputeExporter[T], error) {
  19. pipelineName := pipelines.NameFor[T]()
  20. if pipelineName == "" {
  21. return nil, fmt.Errorf("failed to extract pipeline name for type: %s", typeutil.TypeOf[T]())
  22. }
  23. pathing, err := pathing.NewBingenStoragePathFormatter("federated", clusterId, pipelineName, &resolution)
  24. if err != nil {
  25. return nil, fmt.Errorf("failed to create path formatter: %w", err)
  26. }
  27. return export.NewComputeStorageExporter(
  28. pathing,
  29. export.NewBingenEncoder[T, U](),
  30. store,
  31. validator.NewSetValidator[T, S](resolution),
  32. ), nil
  33. }
  34. // NewComputePipelineExportController creates a new `ComputeExportController[T]` instance which is used to export computed data
  35. // using the provided source, storage, resolution, and source resolution.
  36. func NewComputePipelineExportController[T any, U export.BinaryMarshalerPtr[T], S validator.SetConstraint[T]](
  37. clusterId string,
  38. store storage.Storage,
  39. source export.ComputeSource[T],
  40. resolution time.Duration,
  41. sourceResolution time.Duration,
  42. ) (*export.ComputeExportController[T], error) {
  43. exporter, err := NewComputePipelineExporter[T, U, S](clusterId, resolution, store)
  44. if err != nil {
  45. return nil, fmt.Errorf("failed to create compute exporter: %w", err)
  46. }
  47. return export.NewComputeExportController(source, exporter, resolution, sourceResolution), nil
  48. }