Merge pull request #2258 from opencost/v1.107-patch/cloud-cost-status

Fix cloud status endpoint for aggregator

pkg/cloud/alibaba/boaquerier.go

@@ -25,6 +25,10 @@ type BoaQuerier struct {
 }
 
 func (bq *BoaQuerier) GetStatus() cloud.ConnectionStatus {
+	// initialize status if it has not done so; this can happen if the integration is inactive
+	if bq.ConnectionStatus.String() == "" {
+		bq.ConnectionStatus = cloud.InitialStatus
+	}
 	return bq.ConnectionStatus
 }
 

pkg/cloud/aws/athenaconfiguration.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/opencost/opencost/pkg/cloud/config"
+	"github.com/opencost/opencost/pkg/kubecost"
 	"github.com/opencost/opencost/pkg/util/json"
 )
 
@@ -122,6 +123,10 @@ func (ac *AthenaConfiguration) Key() string {
 	return fmt.Sprintf("%s/%s", ac.Account, ac.Bucket)
 }
 
+func (ac *AthenaConfiguration) Provider() string {
+	return kubecost.AWSProvider
+}
+
 func (ac *AthenaConfiguration) UnmarshalJSON(b []byte) error {
 	var f interface{}
 	err := json.Unmarshal(b, &f)

pkg/cloud/aws/athenaintegration.go

@@ -442,12 +442,3 @@ func (ai *AthenaIntegration) GetConnectionStatusFromResult(result cloud.EmptyChe
 	}
 	return cloud.SuccessfulConnection
 }
-
-func (ai *AthenaIntegration) GetConnectionStatus() string {
-	// initialize status if it has not done so; this can happen if the integration is inactive
-	if ai.ConnectionStatus.String() == "" {
-		ai.ConnectionStatus = cloud.InitialStatus
-	}
-
-	return ai.ConnectionStatus.String()
-}

pkg/cloud/aws/athenaquerier.go

@@ -25,6 +25,10 @@ type AthenaQuerier struct {
 }
 
 func (aq *AthenaQuerier) GetStatus() cloud.ConnectionStatus {
+	// initialize status if it has not done so; this can happen if the integration is inactive
+	if aq.ConnectionStatus.String() == "" {
+		aq.ConnectionStatus = cloud.InitialStatus
+	}
 	return aq.ConnectionStatus
 }
 

pkg/cloud/aws/s3configuration.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/opencost/opencost/pkg/cloud/config"
+	"github.com/opencost/opencost/pkg/kubecost"
 	"github.com/opencost/opencost/pkg/util/json"
 )
 
@@ -89,6 +90,10 @@ func (s3c *S3Configuration) Key() string {
 	return fmt.Sprintf("%s/%s", s3c.Account, s3c.Bucket)
 }
 
+func (s3c *S3Configuration) Provider() string {
+	return kubecost.AWSProvider
+}
+
 func (s3c *S3Configuration) UnmarshalJSON(b []byte) error {
 	var f interface{}
 	err := json.Unmarshal(b, &f)

pkg/cloud/aws/s3connection.go

@@ -15,6 +15,10 @@ type S3Connection struct {
 }
 
 func (s3c *S3Connection) GetStatus() cloud.ConnectionStatus {
+	// initialize status if it has not done so; this can happen if the integration is inactive
+	if s3c.ConnectionStatus.String() == "" {
+		s3c.ConnectionStatus = cloud.InitialStatus
+	}
 	return s3c.ConnectionStatus
 }
 

pkg/cloud/azure/azurestorageintegration.go

@@ -20,7 +20,7 @@ func (asi *AzureStorageIntegration) GetCloudCost(start, end time.Time) (*kubecos
 		return nil, err
 	}
 
-	status, err := asi.ParseBillingData(start, end, func(abv *BillingRowValues) error {
+	err = asi.ParseBillingData(start, end, func(abv *BillingRowValues) error {
 		s := abv.Date
 		e := abv.Date.Add(timeutil.Day)
 		window := kubecost.NewWindow(&s, &e)
@@ -77,7 +77,6 @@ func (asi *AzureStorageIntegration) GetCloudCost(start, end time.Time) (*kubecos
 		return nil
 	})
 	if err != nil {
-		asi.ConnectionStatus = status
 		return nil, err
 	}
 	return ccsr, nil

pkg/cloud/azure/storagebillingparser.go

@@ -30,33 +30,45 @@ func (asbp *AzureStorageBillingParser) Equals(config cloudconfig.Config) bool {
 
 type AzureBillingResultFunc func(*BillingRowValues) error
 
-func (asbp *AzureStorageBillingParser) ParseBillingData(start, end time.Time, resultFn AzureBillingResultFunc) (cloud.ConnectionStatus, error) {
+func (asbp *AzureStorageBillingParser) ParseBillingData(start, end time.Time, resultFn AzureBillingResultFunc) error {
 	err := asbp.Validate()
 	if err != nil {
-		return cloud.InvalidConfiguration, err
+		asbp.ConnectionStatus = cloud.InvalidConfiguration
+		return err
 	}
 
 	containerURL, err := asbp.getContainer()
 	if err != nil {
-		return cloud.FailedConnection, err
+		asbp.ConnectionStatus = cloud.FailedConnection
+		return err
 	}
 	ctx := context.Background()
 	blobNames, err := asbp.getMostRecentBlobs(start, end, containerURL, ctx)
 	if err != nil {
-		return cloud.FailedConnection, err
+		asbp.ConnectionStatus = cloud.FailedConnection
+		return err
+	}
+
+	if len(blobNames) == 0 && asbp.ConnectionStatus != cloud.SuccessfulConnection {
+		asbp.ConnectionStatus = cloud.MissingData
+		return nil
 	}
+
 	for _, blobName := range blobNames {
 		blobBytes, err2 := asbp.DownloadBlob(blobName, containerURL, ctx)
 		if err2 != nil {
-			return cloud.FailedConnection, err2
+			asbp.ConnectionStatus = cloud.FailedConnection
+			return err
 		}
 		err2 = asbp.parseCSV(start, end, csv.NewReader(bytes.NewReader(blobBytes)), resultFn)
 		if err2 != nil {
-			return cloud.ParseError, err2
+			asbp.ConnectionStatus = cloud.ParseError
+			return err2
 		}
 
 	}
-	return cloud.SuccessfulConnection, nil
+	asbp.ConnectionStatus = cloud.SuccessfulConnection
+	return nil
 }
 
 func (asbp *AzureStorageBillingParser) parseCSV(start, end time.Time, reader *csv.Reader, resultFn AzureBillingResultFunc) error {

pkg/cloud/azure/storageconfiguration.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/opencost/opencost/pkg/cloud/config"
+	"github.com/opencost/opencost/pkg/kubecost"
 	"github.com/opencost/opencost/pkg/util/json"
 )
 
@@ -105,6 +106,10 @@ func (sc *StorageConfiguration) Key() string {
 	return key
 }
 
+func (sc *StorageConfiguration) Provider() string {
+	return kubecost.AzureProvider
+}
+
 func (sc *StorageConfiguration) UnmarshalJSON(b []byte) error {
 	var f interface{}
 	err := json.Unmarshal(b, &f)

pkg/cloud/azure/storageconnection.go

@@ -20,6 +20,10 @@ type StorageConnection struct {
 }
 
 func (sc *StorageConnection) GetStatus() cloud.ConnectionStatus {
+	// initialize status if it has not done so; this can happen if the integration is inactive
+	if sc.ConnectionStatus.String() == "" {
+		sc.ConnectionStatus = cloud.InitialStatus
+	}
 	return sc.ConnectionStatus
 }
 

pkg/cloud/cloudcostintegration.go

@@ -9,4 +9,6 @@ import (
 // CloudCostIntegration is an interface for retrieving daily granularity CloudCost data for a given range
 type CloudCostIntegration interface {
 	GetCloudCost(time.Time, time.Time) (*kubecost.CloudCostSetRange, error)
+	Provider() string
+	GetStatus() ConnectionStatus
 }

pkg/cloud/gcp/bigqueryconfiguration.go

@@ -7,6 +7,7 @@ import (
 
 	"cloud.google.com/go/bigquery"
 	"github.com/opencost/opencost/pkg/cloud/config"
+	"github.com/opencost/opencost/pkg/kubecost"
 	"github.com/opencost/opencost/pkg/util/json"
 )
 
@@ -91,6 +92,10 @@ func (bqc *BigQueryConfiguration) Key() string {
 	return fmt.Sprintf("%s/%s", bqc.ProjectID, bqc.GetBillingDataDataset())
 }
 
+func (bqc *BigQueryConfiguration) Provider() string {
+	return kubecost.GCPProvider
+}
+
 func (bqc *BigQueryConfiguration) GetBillingDataDataset() string {
 	return fmt.Sprintf("%s.%s", bqc.Dataset, bqc.Table)
 }

pkg/cloud/gcp/bigqueryquerier.go

@@ -14,6 +14,10 @@ type BigQueryQuerier struct {
 }
 
 func (bqq *BigQueryQuerier) GetStatus() cloud.ConnectionStatus {
+	// initialize status if it has not done so; this can happen if the integration is inactive
+	if bqq.ConnectionStatus.String() == "" {
+		bqq.ConnectionStatus = cloud.InitialStatus
+	}
 	return bqq.ConnectionStatus
 }
 
@@ -41,5 +45,14 @@ func (bqq *BigQueryQuerier) Query(ctx context.Context, queryStr string) (*bigque
 	}
 
 	query := client.Query(queryStr)
-	return query.Read(ctx)
+	iter, err := query.Read(ctx)
+
+	// If result is empty and connection status is not already successful update status to missing data
+	if iter == nil && bqq.ConnectionStatus != cloud.SuccessfulConnection {
+		bqq.ConnectionStatus = cloud.MissingData
+		return iter, nil
+	}
+
+	bqq.ConnectionStatus = cloud.SuccessfulConnection
+	return iter, nil
 }