storageconnection.go 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586
  1. package azure
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "net/url"
  7. "strings"
  8. "github.com/Azure/azure-storage-blob-go/azblob"
  9. "github.com/opencost/opencost/pkg/cloud"
  10. "github.com/opencost/opencost/pkg/log"
  11. )
  12. // StorageConnection provides access to Azure Storage
  13. type StorageConnection struct {
  14. StorageConfiguration
  15. ConnectionStatus cloud.ConnectionStatus
  16. }
  17. func (sc *StorageConnection) GetStatus() cloud.ConnectionStatus {
  18. // initialize status if it has not done so; this can happen if the integration is inactive
  19. if sc.ConnectionStatus.String() == "" {
  20. sc.ConnectionStatus = cloud.InitialStatus
  21. }
  22. return sc.ConnectionStatus
  23. }
  24. func (sc *StorageConnection) Equals(config cloud.Config) bool {
  25. thatConfig, ok := config.(*StorageConnection)
  26. if !ok {
  27. return false
  28. }
  29. return sc.StorageConfiguration.Equals(&thatConfig.StorageConfiguration)
  30. }
  31. func (sc *StorageConnection) getContainer() (*azblob.ContainerURL, error) {
  32. credential, err := sc.Authorizer.GetBlobCredentials()
  33. if err != nil {
  34. return nil, err
  35. }
  36. p := azblob.NewPipeline(credential, azblob.PipelineOptions{})
  37. // From the Azure portal, get your storage account blob service URL endpoint.
  38. URL, _ := url.Parse(
  39. fmt.Sprintf(sc.getBlobURLTemplate(), sc.Account, sc.Container))
  40. // Create a ContainerURL object that wraps the container URL and a request
  41. // pipeline to make requests.
  42. containerURL := azblob.NewContainerURL(*URL, p)
  43. return &containerURL, nil
  44. }
  45. // getBlobURLTemplate returns the correct BlobUrl for whichever Cloud storage account is specified by the AzureCloud configuration
  46. // defaults to the Public Cloud template
  47. func (sc *StorageConnection) getBlobURLTemplate() string {
  48. // Use gov cloud blob url if gov is detected in AzureCloud
  49. if strings.Contains(strings.ToLower(sc.Cloud), "gov") {
  50. return "https://%s.blob.core.usgovcloudapi.net/%s"
  51. }
  52. // default to Public Cloud template
  53. return "https://%s.blob.core.windows.net/%s"
  54. }
  55. func (sc *StorageConnection) DownloadBlob(blobName string, containerURL *azblob.ContainerURL, ctx context.Context) ([]byte, error) {
  56. log.Infof("Azure Storage: retrieving blob: %v", blobName)
  57. blobURL := containerURL.NewBlobURL(blobName)
  58. downloadResponse, err := blobURL.Download(ctx, 0, azblob.CountToEnd, azblob.BlobAccessConditions{}, false, azblob.ClientProvidedKeyOptions{})
  59. if err != nil {
  60. return nil, err
  61. }
  62. // NOTE: automatically retries are performed if the connection fails
  63. bodyStream := downloadResponse.Body(azblob.RetryReaderOptions{MaxRetryRequests: 20})
  64. // read the body into a buffer
  65. downloadedData := bytes.Buffer{}
  66. _, err = downloadedData.ReadFrom(bodyStream)
  67. if err != nil {
  68. return nil, err
  69. }
  70. return downloadedData.Bytes(), nil
  71. }