#3842

Committed 07 Apr 2025 11:21AM UTC coverage: 62.696% (-0.3%) from 63.027%

Build # #3842

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch (#30679)

Run Details

154855 of 315075 branches covered (49.15%)

Branch coverage included in aggregate %.

6 of 8 new or added lines in 5 files covered. (75.0%)

2309 existing lines in 130 files now uncovered.

240176 of 314995 relevant lines covered (76.25%)

19119980.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

46.07

/source/libs/azure/src/az.cpp

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#define ALLOW_FORBID_FUNC

#include "az.h"
#include "azInt.h"

#include "os.h"
#include "taoserror.h"
#include "tglobal.h"

int32_t azBegin() { return TSDB_CODE_SUCCESS; }

void azEnd() {}

#if defined(USE_S3)

#include <azure/core.hpp>
#include <azure/storage/blobs.hpp>
#include "td_block_blob_client.hpp"

using namespace Azure::Storage;
using namespace Azure::Storage::Blobs;

extern char tsS3Hostname[][TSDB_FQDN_LEN];
extern char tsS3AccessKeyId[][TSDB_FQDN_LEN];
extern char tsS3AccessKeySecret[][TSDB_FQDN_LEN];
extern char tsS3BucketName[TSDB_FQDN_LEN];

extern int8_t tsS3Enabled;
extern int8_t tsS3EpNum;

static void checkPrint(const char *fmt, ...) {
  va_list arg_ptr;
  va_start(arg_ptr, fmt);
  (void)vfprintf(stderr, fmt, arg_ptr);
  va_end(arg_ptr);
}

static void azDumpCfgByEp(int8_t epIndex) {
  // clang-format off
  checkPrint(
      "%-24s %s\n"
      "%-24s %s\n"
      "%-24s %s\n"
      "%-24s %s\n"
      "%-24s %s\n"
      "%-24s %s\n",
      "hostName", tsS3Hostname[epIndex],
      "bucketName", tsS3BucketName,
      "protocol", "https only",
      "uristyle", "path only",
      "accessKey", tsS3AccessKeyId[epIndex],
      "accessKeySecret", tsS3AccessKeySecret[epIndex]);
  // clang-format on
}

static int32_t azListBucket(char const *bucketname) {
  int32_t           code = 0;
  const std::string delimiter = "/";
  std::string       accountName = tsS3AccessKeyId[0];
  std::string       accountKey = tsS3AccessKeySecret[0];
  std::string       accountURL = tsS3Hostname[0];
  accountURL = "https://" + accountURL;

  try {
    auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);

    BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);

    std::string containerName = bucketname;
    auto        containerClient = blobServiceClient.GetBlobContainerClient(containerName);

    Azure::Storage::Blobs::ListBlobsOptions options;
    options.Prefix = "s3";

    checkPrint("objects:\n");
    for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {
      for (const auto &blob : pageResult.Blobs) {
        checkPrint("%s\n", blob.Name.c_str());
      }
    }
  } catch (const Azure::Core::RequestFailedException &e) {
    azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),
            e.ReasonPhrase.c_str());

    code = TAOS_SYSTEM_ERROR(EIO);
    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

int32_t azCheckCfg() {
  int32_t code = 0, lino = 0;

  azDumpCfgByEp(0);

  // test put
  char        testdata[17] = "0123456789abcdef";
  const char *objectname[] = {"s3test.txt"};
  char        path[PATH_MAX] = {0};
  int         ds_len = strlen(TD_DIRSEP);
  int         tmp_len = strlen(tsTempDir);

  (void)snprintf(path, PATH_MAX, "%s", tsTempDir);
  if (strncmp(tsTempDir + tmp_len - ds_len, TD_DIRSEP, ds_len) != 0) {
    (void)snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", TD_DIRSEP);
    (void)snprintf(path + tmp_len + ds_len, PATH_MAX - tmp_len - ds_len, "%s", objectname[0]);
  } else {
    (void)snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", objectname[0]);
  }

  uint8_t *pBlock = NULL;
  int      c_offset = 10;
  int      c_len = 6;
  char     buf[7] = {0};

  TdFilePtr fp = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
  if (!fp) {
    checkPrint("failed to open test file: %s.\n", path);
    TAOS_CHECK_GOTO(terrno, &lino, _next);
  }
  if (taosWriteFile(fp, testdata, strlen(testdata)) < 0) {
    checkPrint("failed to write test file: %s.\n", path);
    TAOS_CHECK_GOTO(terrno, &lino, _next);
  }
  if (taosFsyncFile(fp) < 0) {
    checkPrint("failed to fsync test file: %s.\n", path);
    TAOS_CHECK_GOTO(terrno, &lino, _next);
  }
  (void)taosCloseFile(&fp);

  checkPrint("\nstart to put object: %s, file: %s content: %s\n", objectname[0], path, testdata);
  code = azPutObjectFromFileOffset(path, objectname[0], 0, 16);
  if (code != 0) {
    checkPrint("put object %s : failed.\n", objectname[0]);
    TAOS_CHECK_GOTO(code, &lino, _next);
  }
  checkPrint("put object %s: success.\n\n", objectname[0]);

  // list buckets
  checkPrint("start to list bucket %s by prefix s3.\n", tsS3BucketName);
  code = azListBucket(tsS3BucketName);
  if (code != 0) {
    checkPrint("listing bucket %s : failed.\n", tsS3BucketName);
    TAOS_CHECK_GOTO(code, &lino, _next);
  }
  checkPrint("listing bucket %s: success.\n\n", tsS3BucketName);

  // test range get
  checkPrint("start to range get object %s offset: %d len: %d.\n", objectname[0], c_offset, c_len);
  code = azGetObjectBlock(objectname[0], c_offset, c_len, true, &pBlock);
  if (code != 0) {
    checkPrint("get object %s : failed.\n", objectname[0]);
    TAOS_CHECK_GOTO(code, &lino, _next);
  }

  (void)memcpy(buf, pBlock, c_len);
  taosMemoryFree(pBlock);
  checkPrint("object content: %s\n", buf);
  checkPrint("get object %s: success.\n\n", objectname[0]);

  // delete test object
  checkPrint("start to delete object: %s.\n", objectname[0]);
  // code = azDeleteObjectsByPrefix(objectname[0]);
  azDeleteObjectsByPrefix(objectname[0]);
  /*
  if (code != 0) {
    (void)fprintf(stderr, "delete object %s : failed.\n", objectname[0]);
    TAOS_CHECK_GOTO(code, &lino, _next);
  }
  */
  checkPrint("delete object %s: success.\n\n", objectname[0]);

_next:
  if (fp) {
    (void)taosCloseFile(&fp);
  }

  if (TSDB_CODE_SUCCESS != code) {
    checkPrint("s3 check failed, code: %d, line: %d.\n", code, lino);
  }

  checkPrint("=================================================================\n");

  TAOS_RETURN(code);
}

static int32_t azPutObjectFromFileOffsetImpl(const char *file, const char *object_name, int64_t offset, int64_t size) {
  int32_t code = 0;

  std::string endpointUrl = tsS3Hostname[0];
  std::string accountName = tsS3AccessKeyId[0];
  std::string accountKey = tsS3AccessKeySecret[0];

  try {
    auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);

    std::string accountURL = tsS3Hostname[0];

    accountURL = "https://" + accountURL;
    BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);

    std::string containerName = tsS3BucketName;
    auto        containerClient = blobServiceClient.GetBlobContainerClient(containerName);

    std::string blobName = "blob.txt";
    uint8_t     blobContent[] = "Hello Azure!";
    // Create the block blob client
    // BlockBlobClient blobClient = containerClient.GetBlockBlobClient(blobName);
    TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));

    blobClient.UploadFrom(file, offset, size);
  } catch (const Azure::Core::RequestFailedException &e) {
    azError("%s: Status Code: %d, Reason Phrase: %s", __func__, static_cast<int>(e.StatusCode), e.ReasonPhrase.c_str());

    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

int32_t azPutObjectFromFileOffset(const char *file, const char *object_name, int64_t offset, int64_t size) {
  int32_t code = 0;

  try {
    code = azPutObjectFromFileOffsetImpl(file, object_name, offset, size);
  } catch (const std::exception &e) {
    azError("%s: Reason Phrase: %s", __func__, e.what());

    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

static int32_t azGetObjectBlockImpl(const char *object_name, int64_t offset, int64_t size, bool check,
                                    uint8_t **ppBlock) {
  int32_t     code = TSDB_CODE_SUCCESS;
  std::string accountName = tsS3AccessKeyId[0];
  std::string accountKey = tsS3AccessKeySecret[0];
  std::string accountURL = tsS3Hostname[0];
  uint8_t    *buf = NULL;

  try {
    auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);

    accountURL = "https://" + accountURL;
    BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);

    std::string containerName = tsS3BucketName;
    auto        containerClient = blobServiceClient.GetBlobContainerClient(containerName);

    TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));

    Blobs::DownloadBlobToOptions options;
    options.Range = Azure::Core::Http::HttpRange();
    options.Range.Value().Offset = offset;
    options.Range.Value().Length = size;

    buf = (uint8_t *)taosMemoryCalloc(1, size);
    if (!buf) {
      return terrno;
    }

    auto res = blobClient.DownloadTo(buf, size, options);
    if (check && res.Value.ContentRange.Length.Value() != size) {
      code = TAOS_SYSTEM_ERROR(EIO);
      azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
      TAOS_RETURN(code);
    }

    *ppBlock = buf;
  } catch (const Azure::Core::RequestFailedException &e) {
    azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),
            e.ReasonPhrase.c_str());
    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

    if (buf) {
      taosMemoryFree(buf);
    }
    *ppBlock = NULL;

    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

static int32_t azGetObjectBlockRetry(const char *object_name, int64_t offset, int64_t size, bool check,
                                     uint8_t **ppBlock) {
  int32_t code = TSDB_CODE_SUCCESS;

  // May use an exponential backoff policy for retries with 503
  int        retryCount = 0;
  static int maxRetryCount = 5;
  static int minRetryInterval = 1000;  // ms
  static int maxRetryInterval = 3000;  // ms

_retry:
  code = azGetObjectBlockImpl(object_name, offset, size, check, ppBlock);
  if (TSDB_CODE_SUCCESS != code && retryCount++ < maxRetryCount) {
    taosMsleep(taosRand() % (maxRetryInterval - minRetryInterval + 1) + minRetryInterval);
    uInfo("%s: 0x%x(%s) and retry get object", __func__, code, tstrerror(code));
    goto _retry;
  }

  TAOS_RETURN(code);
}

int32_t azGetObjectBlock(const char *object_name, int64_t offset, int64_t size, bool check, uint8_t **ppBlock) {
  int32_t code = TSDB_CODE_SUCCESS;

  try {
    code = azGetObjectBlockRetry(object_name, offset, size, check, ppBlock);
  } catch (const std::exception &e) {
    azError("%s: Reason Phrase: %s", __func__, e.what());

    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

static void azDeleteObjectsByPrefixImpl(const char *prefix) {
  const std::string delimiter = "/";
  std::string       accountName = tsS3AccessKeyId[0];
  std::string       accountKey = tsS3AccessKeySecret[0];
  std::string       accountURL = tsS3Hostname[0];
  accountURL = "https://" + accountURL;

  try {
    auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);

    BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);

    std::string containerName = tsS3BucketName;
    auto        containerClient = blobServiceClient.GetBlobContainerClient(containerName);

    Azure::Storage::Blobs::ListBlobsOptions options;
    options.Prefix = prefix;

    std::set<std::string> listBlobs;
    for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {
      for (const auto &blob : pageResult.Blobs) {
        listBlobs.insert(blob.Name);
      }
    }

    for (auto blobName : listBlobs) {
      auto blobClient = containerClient.GetAppendBlobClient(blobName);
      blobClient.Delete();
    }
  } catch (const Azure::Core::RequestFailedException &e) {
    azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),
            e.ReasonPhrase.c_str());
  }
}

void azDeleteObjectsByPrefix(const char *prefix) {
  int32_t code = TSDB_CODE_SUCCESS;

  try {
    azDeleteObjectsByPrefixImpl(prefix);
  } catch (const std::exception &e) {
    azError("%s: Reason Phrase: %s", __func__, e.what());
  }
}

int32_t azPutObjectFromFile2(const char *file, const char *object, int8_t withcp) {
  int32_t  code = 0, lino = 0;
  uint64_t contentLength = 0;

  if (taosStatFile(file, (int64_t *)&contentLength, NULL, NULL) < 0) {
    azError("ERROR: %s Failed to stat file %s: ", __func__, file);
    TAOS_RETURN(terrno);
  }

  code = azPutObjectFromFileOffset(file, object, 0, contentLength);
  if (code != 0) {
    azError("ERROR: %s Failed to put file %s: ", __func__, file);
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }

_exit:
  if (code) {
    azError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }

  return 0;
}

static int32_t azGetObjectToFileImpl(const char *object_name, const char *fileName) {
  int32_t     code = TSDB_CODE_SUCCESS;
  std::string accountName = tsS3AccessKeyId[0];
  std::string accountKey = tsS3AccessKeySecret[0];
  std::string accountURL = tsS3Hostname[0];
  accountURL = "https://" + accountURL;

  try {
    auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);

    BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);

    std::string containerName = tsS3BucketName;
    auto        containerClient = blobServiceClient.GetBlobContainerClient(containerName);

    TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));

    auto res = blobClient.DownloadTo(fileName);
    if (res.Value.ContentRange.Length.Value() <= 0) {
      code = TAOS_SYSTEM_ERROR(EIO);
      azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
      TAOS_RETURN(code);
    }
  } catch (const Azure::Core::RequestFailedException &e) {
    azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),
            e.ReasonPhrase.c_str());
    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

int32_t azGetObjectToFile(const char *object_name, const char *fileName) {
  int32_t code = 0;

  try {
    code = azGetObjectToFileImpl(object_name, fileName);
  } catch (const std::exception &e) {
    azError("%s: Reason Phrase: %s", __func__, e.what());

    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

static int32_t azGetObjectsByPrefixImpl(const char *prefix, const char *path) {
  const std::string delimiter = "/";
  std::string       accountName = tsS3AccessKeyId[0];
  std::string       accountKey = tsS3AccessKeySecret[0];
  std::string       accountURL = tsS3Hostname[0];
  accountURL = "https://" + accountURL;

  try {
    auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);

    BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);

    std::string containerName = tsS3BucketName;
    auto        containerClient = blobServiceClient.GetBlobContainerClient(containerName);

    Azure::Storage::Blobs::ListBlobsOptions options;
    options.Prefix = prefix;

    std::set<std::string> listBlobs;
    for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {
      for (const auto &blob : pageResult.Blobs) {
        listBlobs.insert(blob.Name);
      }
    }

    for (auto blobName : listBlobs) {
      const char *tmp = strchr(blobName.c_str(), '/');
      tmp = (tmp == NULL) ? blobName.c_str() : tmp + 1;
      char fileName[PATH_MAX] = {0};
      if (path[strlen(path) - 1] != TD_DIRSEP_CHAR) {
        (void)snprintf(fileName, PATH_MAX, "%s%s%s", path, TD_DIRSEP, tmp);
      } else {
        (void)snprintf(fileName, PATH_MAX, "%s%s", path, tmp);
      }
      if (azGetObjectToFile(blobName.c_str(), fileName)) {
        TAOS_RETURN(TSDB_CODE_FAILED);
      }
    }
  } catch (const Azure::Core::RequestFailedException &e) {
    azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),
            e.ReasonPhrase.c_str());
    TAOS_RETURN(TSDB_CODE_FAILED);
  }

  return 0;
}

int32_t azGetObjectsByPrefix(const char *prefix, const char *path) {
  int32_t code = 0;

  try {
    code = azGetObjectsByPrefixImpl(prefix, path);
  } catch (const std::exception &e) {
    azError("%s: Reason Phrase: %s", __func__, e.what());

    code = TAOS_SYSTEM_ERROR(EIO);
    azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

    TAOS_RETURN(code);
  }

  TAOS_RETURN(code);
}

int32_t azDeleteObjects(const char *object_name[], int nobject) {
  for (int i = 0; i < nobject; ++i) {
    azDeleteObjectsByPrefix(object_name[i]);
  }

  return 0;
}

#else

int32_t azCheckCfg() { return TSDB_CODE_SUCCESS; }

int32_t azPutObjectFromFileOffset(const char *file, const char *object_name, int64_t offset, int64_t size) {
  return TSDB_CODE_SUCCESS;
}

int32_t azGetObjectBlock(const char *object_name, int64_t offset, int64_t size, bool check, uint8_t **ppBlock) {
  return TSDB_CODE_SUCCESS;
}

void azDeleteObjectsByPrefix(const char *prefix) {}

int32_t azPutObjectFromFile2(const char *file, const char *object, int8_t withcp) { return 0; }

int32_t azGetObjectsByPrefix(const char *prefix, const char *path) { return 0; }

int32_t azGetObjectToFile(const char *object_name, const char *fileName) { return 0; }

int32_t azDeleteObjects(const char *object_name[], int nobject) { return 0; }

#endif

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define ALLOW_FORBID_FUNC
17
18	#include "az.h"
19	#include "azInt.h"
20
21	#include "os.h"
22	#include "taoserror.h"
23	#include "tglobal.h"
24
25	int32_t azBegin() { return TSDB_CODE_SUCCESS; }	3✔
26
27	void azEnd() {}	3✔
28
29	#if defined(USE_S3)
30
31	#include <azure/core.hpp>
32	#include <azure/storage/blobs.hpp>
33	#include "td_block_blob_client.hpp"
34
35	using namespace Azure::Storage;
36	using namespace Azure::Storage::Blobs;
37
38	extern char tsS3Hostname[][TSDB_FQDN_LEN];
39	extern char tsS3AccessKeyId[][TSDB_FQDN_LEN];
40	extern char tsS3AccessKeySecret[][TSDB_FQDN_LEN];
41	extern char tsS3BucketName[TSDB_FQDN_LEN];
42
43	extern int8_t tsS3Enabled;
44	extern int8_t tsS3EpNum;
45
46	static void checkPrint(const char *fmt, ...) {	26✔
47	va_list arg_ptr;
48	va_start(arg_ptr, fmt);	26✔
49	(void)vfprintf(stderr, fmt, arg_ptr);	26!
50	va_end(arg_ptr);	26✔
51	}	26✔
52
53	static void azDumpCfgByEp(int8_t epIndex) {	2✔
54	// clang-format off
55	checkPrint(	2✔
56	"%-24s %s\n"
57	"%-24s %s\n"
58	"%-24s %s\n"
59	"%-24s %s\n"
60	"%-24s %s\n"
61	"%-24s %s\n",
62	"hostName", tsS3Hostname[epIndex],	2✔
63	"bucketName", tsS3BucketName,
64	"protocol", "https only",
65	"uristyle", "path only",
66	"accessKey", tsS3AccessKeyId[epIndex],	2✔
67	"accessKeySecret", tsS3AccessKeySecret[epIndex]);	2✔
68	// clang-format on
69	}	2✔
70
71	static int32_t azListBucket(char const *bucketname) {	2✔
72	int32_t code = 0;	2✔
73	const std::string delimiter = "/";	2!
74	std::string accountName = tsS3AccessKeyId[0];	2!
75	std::string accountKey = tsS3AccessKeySecret[0];	2!
76	std::string accountURL = tsS3Hostname[0];	2!
77	accountURL = "https://" + accountURL;	2!
78
79	try {
80	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	2!
81
82	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	4!
83
84	std::string containerName = bucketname;	2!
85	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	2!
86
87	Azure::Storage::Blobs::ListBlobsOptions options;	2✔
88	options.Prefix = "s3";	2!
89
90	checkPrint("objects:\n");	2!
91	for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {	6!
92	for (const auto &blob : pageResult.Blobs) {	4✔
93	checkPrint("%s\n", blob.Name.c_str());	2!
94	}
95	}	2✔
96	} catch (const Azure::Core::RequestFailedException &e) {	2!
97	azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),	×
98	e.ReasonPhrase.c_str());
99
100	code = TAOS_SYSTEM_ERROR(EIO);	×
101	TAOS_RETURN(code);	×
102	}	×
103
104	TAOS_RETURN(code);	2!
105	}	2✔
106
107	int32_t azCheckCfg() {	2✔
108	int32_t code = 0, lino = 0;	2✔
109
110	azDumpCfgByEp(0);	2!
111
112	// test put
113	char testdata[17] = "0123456789abcdef";	2✔
114	const char *objectname[] = {"s3test.txt"};	2✔
115	char path[PATH_MAX] = {0};	2✔
116	int ds_len = strlen(TD_DIRSEP);	2✔
117	int tmp_len = strlen(tsTempDir);	2✔
118
119	(void)snprintf(path, PATH_MAX, "%s", tsTempDir);	2✔
120	if (strncmp(tsTempDir + tmp_len - ds_len, TD_DIRSEP, ds_len) != 0) {	2!
121	(void)snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", TD_DIRSEP);	×
122	(void)snprintf(path + tmp_len + ds_len, PATH_MAX - tmp_len - ds_len, "%s", objectname[0]);	×
123	} else {
124	(void)snprintf(path + tmp_len, PATH_MAX - tmp_len, "%s", objectname[0]);	2✔
125	}
126
127	uint8_t *pBlock = NULL;	2✔
128	int c_offset = 10;	2✔
129	int c_len = 6;	2✔
130	char buf[7] = {0};	2✔
131
132	TdFilePtr fp = taosOpenFile(path, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_READ \| TD_FILE_TRUNC);	2!
133	if (!fp) {	2!
134	checkPrint("failed to open test file: %s.\n", path);	×
135	TAOS_CHECK_GOTO(terrno, &lino, _next);	×
136	}
137	if (taosWriteFile(fp, testdata, strlen(testdata)) < 0) {	2!
138	checkPrint("failed to write test file: %s.\n", path);	×
139	TAOS_CHECK_GOTO(terrno, &lino, _next);	×
140	}
141	if (taosFsyncFile(fp) < 0) {	2!
142	checkPrint("failed to fsync test file: %s.\n", path);	×
143	TAOS_CHECK_GOTO(terrno, &lino, _next);	×
144	}
145	(void)taosCloseFile(&fp);	2!
146
147	checkPrint("\nstart to put object: %s, file: %s content: %s\n", objectname[0], path, testdata);	2!
148	code = azPutObjectFromFileOffset(path, objectname[0], 0, 16);	2!
149	if (code != 0) {	2!
150	checkPrint("put object %s : failed.\n", objectname[0]);	×
151	TAOS_CHECK_GOTO(code, &lino, _next);	×
152	}
153	checkPrint("put object %s: success.\n\n", objectname[0]);	2!
154
155	// list buckets
156	checkPrint("start to list bucket %s by prefix s3.\n", tsS3BucketName);	2!
157	code = azListBucket(tsS3BucketName);	2!
158	if (code != 0) {	2!
159	checkPrint("listing bucket %s : failed.\n", tsS3BucketName);	×
160	TAOS_CHECK_GOTO(code, &lino, _next);	×
161	}
162	checkPrint("listing bucket %s: success.\n\n", tsS3BucketName);	2!
163
164	// test range get
165	checkPrint("start to range get object %s offset: %d len: %d.\n", objectname[0], c_offset, c_len);	2!
166	code = azGetObjectBlock(objectname[0], c_offset, c_len, true, &pBlock);	2!
167	if (code != 0) {	2!
168	checkPrint("get object %s : failed.\n", objectname[0]);	×
169	TAOS_CHECK_GOTO(code, &lino, _next);	×
170	}
171
172	(void)memcpy(buf, pBlock, c_len);	2✔
173	taosMemoryFree(pBlock);	2!
174	checkPrint("object content: %s\n", buf);	2!
175	checkPrint("get object %s: success.\n\n", objectname[0]);	2!
176
177	// delete test object
178	checkPrint("start to delete object: %s.\n", objectname[0]);	2!
179	// code = azDeleteObjectsByPrefix(objectname[0]);
180	azDeleteObjectsByPrefix(objectname[0]);	2!
181	/*
182	if (code != 0) {
183	(void)fprintf(stderr, "delete object %s : failed.\n", objectname[0]);
184	TAOS_CHECK_GOTO(code, &lino, _next);
185	}
186	*/
187	checkPrint("delete object %s: success.\n\n", objectname[0]);	2!
188
189	_next:	2✔
190	if (fp) {	2!
191	(void)taosCloseFile(&fp);	×
192	}
193
194	if (TSDB_CODE_SUCCESS != code) {	2!
195	checkPrint("s3 check failed, code: %d, line: %d.\n", code, lino);	×
196	}
197
198	checkPrint("=================================================================\n");	2!
199
200	TAOS_RETURN(code);	2!
201	}
202
203	static int32_t azPutObjectFromFileOffsetImpl(const char file, const char object_name, int64_t offset, int64_t size) {	4✔
204	int32_t code = 0;	4✔
205
206	std::string endpointUrl = tsS3Hostname[0];	4!
207	std::string accountName = tsS3AccessKeyId[0];	4!
208	std::string accountKey = tsS3AccessKeySecret[0];	4!
209
210	try {
211	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	4!
212
213	std::string accountURL = tsS3Hostname[0];	4!
214
215	accountURL = "https://" + accountURL;	4!
216	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	8!
217
218	std::string containerName = tsS3BucketName;	4!
219	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	4!
220
221	std::string blobName = "blob.txt";	4!
222	uint8_t blobContent[] = "Hello Azure!";	4✔
223	// Create the block blob client
224	// BlockBlobClient blobClient = containerClient.GetBlockBlobClient(blobName);
225	TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));	8!
226
227	blobClient.UploadFrom(file, offset, size);	4!
228	} catch (const Azure::Core::RequestFailedException &e) {	4!
229	azError("%s: Status Code: %d, Reason Phrase: %s", __func__, static_cast<int>(e.StatusCode), e.ReasonPhrase.c_str());	×
230
231	code = TAOS_SYSTEM_ERROR(EIO);	×
232	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
233
234	TAOS_RETURN(code);	×
235	}	×
236
237	TAOS_RETURN(code);	4!
238	}	4✔
239
240	int32_t azPutObjectFromFileOffset(const char file, const char object_name, int64_t offset, int64_t size) {	4✔
241	int32_t code = 0;	4✔
242
243	try {
244	code = azPutObjectFromFileOffsetImpl(file, object_name, offset, size);	4!
245	} catch (const std::exception &e) {	×
246	azError("%s: Reason Phrase: %s", __func__, e.what());	×
247
248	code = TAOS_SYSTEM_ERROR(EIO);	×
249	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
250
251	TAOS_RETURN(code);	×
252	}	×
253
254	TAOS_RETURN(code);	4✔
255	}
256
257	static int32_t azGetObjectBlockImpl(const char *object_name, int64_t offset, int64_t size, bool check,	3✔
258	uint8_t **ppBlock) {
259	int32_t code = TSDB_CODE_SUCCESS;	3✔
260	std::string accountName = tsS3AccessKeyId[0];	3!
261	std::string accountKey = tsS3AccessKeySecret[0];	3!
262	std::string accountURL = tsS3Hostname[0];	3!
263	uint8_t *buf = NULL;	3✔
264
265	try {
266	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	3!
267
268	accountURL = "https://" + accountURL;	3!
269	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	6!
270
271	std::string containerName = tsS3BucketName;	3!
272	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	3!
273
274	TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));	6!
275
276	Blobs::DownloadBlobToOptions options;	3✔
277	options.Range = Azure::Core::Http::HttpRange();	3✔
278	options.Range.Value().Offset = offset;	3✔
279	options.Range.Value().Length = size;	3✔
280
281	buf = (uint8_t *)taosMemoryCalloc(1, size);	3!
282	if (!buf) {	3!
283	return terrno;	×
284	}
285
286	auto res = blobClient.DownloadTo(buf, size, options);	3!
287	if (check && res.Value.ContentRange.Length.Value() != size) {	3!
288	code = TAOS_SYSTEM_ERROR(EIO);	×
289	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
290	TAOS_RETURN(code);	×
291	}
292
293	*ppBlock = buf;	3✔
294	} catch (const Azure::Core::RequestFailedException &e) {	3!
295	azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),	×
296	e.ReasonPhrase.c_str());
297	code = TAOS_SYSTEM_ERROR(EIO);	×
298	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
299
300	if (buf) {	×
301	taosMemoryFree(buf);	×
302	}
303	*ppBlock = NULL;	×
304
305	TAOS_RETURN(code);	×
306	}	×
307
308	TAOS_RETURN(code);	3!
309	}	3✔
310
311	static int32_t azGetObjectBlockRetry(const char *object_name, int64_t offset, int64_t size, bool check,	3✔
312	uint8_t **ppBlock) {
313	int32_t code = TSDB_CODE_SUCCESS;	3✔
314
315	// May use an exponential backoff policy for retries with 503
316	int retryCount = 0;	3✔
317	static int maxRetryCount = 5;
318	static int minRetryInterval = 1000; // ms
319	static int maxRetryInterval = 3000; // ms
320
321	_retry:	3✔
322	code = azGetObjectBlockImpl(object_name, offset, size, check, ppBlock);	3✔
323	if (TSDB_CODE_SUCCESS != code && retryCount++ < maxRetryCount) {	3!
324	taosMsleep(taosRand() % (maxRetryInterval - minRetryInterval + 1) + minRetryInterval);	×
325	uInfo("%s: 0x%x(%s) and retry get object", __func__, code, tstrerror(code));	×
326	goto _retry;	×
327	}
328
329	TAOS_RETURN(code);	3✔
330	}
331
332	int32_t azGetObjectBlock(const char object_name, int64_t offset, int64_t size, bool check, uint8_t *ppBlock) {	3✔
333	int32_t code = TSDB_CODE_SUCCESS;	3✔
334
335	try {
336	code = azGetObjectBlockRetry(object_name, offset, size, check, ppBlock);	3!
337	} catch (const std::exception &e) {	×
338	azError("%s: Reason Phrase: %s", __func__, e.what());	×
339
340	code = TAOS_SYSTEM_ERROR(EIO);	×
341	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
342
343	TAOS_RETURN(code);	×
344	}	×
345
346	TAOS_RETURN(code);	3✔
347	}
348
349	static void azDeleteObjectsByPrefixImpl(const char *prefix) {	4✔
350	const std::string delimiter = "/";	4!
351	std::string accountName = tsS3AccessKeyId[0];	4!
352	std::string accountKey = tsS3AccessKeySecret[0];	4!
353	std::string accountURL = tsS3Hostname[0];	4!
354	accountURL = "https://" + accountURL;	4!
355
356	try {
357	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	4!
358
359	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	8!
360
361	std::string containerName = tsS3BucketName;	4!
362	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	4!
363
364	Azure::Storage::Blobs::ListBlobsOptions options;	4✔
365	options.Prefix = prefix;	4!
366
367	std::set<std::string> listBlobs;	4✔
368	for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {	12!
369	for (const auto &blob : pageResult.Blobs) {	7✔
370	listBlobs.insert(blob.Name);	3!
371	}
372	}	4✔
373
374	for (auto blobName : listBlobs) {	7!
375	auto blobClient = containerClient.GetAppendBlobClient(blobName);	3!
376	blobClient.Delete();	3!
377	}	3✔
378	} catch (const Azure::Core::RequestFailedException &e) {	4!
UNCOV 379	azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),	×
380	e.ReasonPhrase.c_str());
UNCOV 381	}	×
382	}	4✔
383
384	void azDeleteObjectsByPrefix(const char *prefix) {	4✔
385	int32_t code = TSDB_CODE_SUCCESS;	4✔
386
387	try {
388	azDeleteObjectsByPrefixImpl(prefix);	4!
389	} catch (const std::exception &e) {	×
390	azError("%s: Reason Phrase: %s", __func__, e.what());	×
391	}	×
392	}	4✔
393
394	int32_t azPutObjectFromFile2(const char file, const char object, int8_t withcp) {	1✔
395	int32_t code = 0, lino = 0;	1✔
396	uint64_t contentLength = 0;	1✔
397
398	if (taosStatFile(file, (int64_t *)&contentLength, NULL, NULL) < 0) {	1!
399	azError("ERROR: %s Failed to stat file %s: ", __func__, file);	×
400	TAOS_RETURN(terrno);	×
401	}
402
403	code = azPutObjectFromFileOffset(file, object, 0, contentLength);	1!
404	if (code != 0) {	1!
405	azError("ERROR: %s Failed to put file %s: ", __func__, file);	×
406	TAOS_CHECK_GOTO(code, &lino, _exit);	×
407	}
408
409	_exit:	1✔
410	if (code) {	1!
411	azError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
412	}
413
414	return 0;	1✔
415	}
416
417	static int32_t azGetObjectToFileImpl(const char object_name, const char fileName) {	2✔
418	int32_t code = TSDB_CODE_SUCCESS;	2✔
419	std::string accountName = tsS3AccessKeyId[0];	2!
420	std::string accountKey = tsS3AccessKeySecret[0];	2!
421	std::string accountURL = tsS3Hostname[0];	2!
422	accountURL = "https://" + accountURL;	2!
423
424	try {
425	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	2!
426
427	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	4!
428
429	std::string containerName = tsS3BucketName;	2!
430	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	2!
431
432	TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));	4!
433
434	auto res = blobClient.DownloadTo(fileName);	4!
435	if (res.Value.ContentRange.Length.Value() <= 0) {	2!
436	code = TAOS_SYSTEM_ERROR(EIO);	×
437	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
438	TAOS_RETURN(code);	×
439	}
440	} catch (const Azure::Core::RequestFailedException &e) {	2!
441	azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),	×
442	e.ReasonPhrase.c_str());
443	code = TAOS_SYSTEM_ERROR(EIO);	×
444	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
445	TAOS_RETURN(code);	×
446	}	×
447
448	TAOS_RETURN(code);	2!
449	}	2✔
450
451	int32_t azGetObjectToFile(const char object_name, const char fileName) {	2✔
452	int32_t code = 0;	2✔
453
454	try {
455	code = azGetObjectToFileImpl(object_name, fileName);	2!
456	} catch (const std::exception &e) {	×
457	azError("%s: Reason Phrase: %s", __func__, e.what());	×
458
459	code = TAOS_SYSTEM_ERROR(EIO);	×
460	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
461
462	TAOS_RETURN(code);	×
463	}	×
464
465	TAOS_RETURN(code);	2✔
466	}
467
468	static int32_t azGetObjectsByPrefixImpl(const char prefix, const char path) {	1✔
469	const std::string delimiter = "/";	1!
470	std::string accountName = tsS3AccessKeyId[0];	1!
471	std::string accountKey = tsS3AccessKeySecret[0];	1!
472	std::string accountURL = tsS3Hostname[0];	1!
473	accountURL = "https://" + accountURL;	1!
474
475	try {
476	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	1!
477
478	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	2!
479
480	std::string containerName = tsS3BucketName;	1!
481	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	1!
482
483	Azure::Storage::Blobs::ListBlobsOptions options;	1✔
484	options.Prefix = prefix;	1!
485
486	std::set<std::string> listBlobs;	1✔
487	for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {	3!
488	for (const auto &blob : pageResult.Blobs) {	2✔
489	listBlobs.insert(blob.Name);	1!
490	}
491	}	1✔
492
493	for (auto blobName : listBlobs) {	2!
494	const char *tmp = strchr(blobName.c_str(), '/');	1✔
495	tmp = (tmp == NULL) ? blobName.c_str() : tmp + 1;	1!
496	char fileName[PATH_MAX] = {0};	1✔
497	if (path[strlen(path) - 1] != TD_DIRSEP_CHAR) {	1!
498	(void)snprintf(fileName, PATH_MAX, "%s%s%s", path, TD_DIRSEP, tmp);	×
499	} else {
500	(void)snprintf(fileName, PATH_MAX, "%s%s", path, tmp);	1✔
501	}
502	if (azGetObjectToFile(blobName.c_str(), fileName)) {	1!
503	TAOS_RETURN(TSDB_CODE_FAILED);	×
504	}
505	}	1!
506	} catch (const Azure::Core::RequestFailedException &e) {	1!
507	azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),	×
508	e.ReasonPhrase.c_str());
509	TAOS_RETURN(TSDB_CODE_FAILED);	×
510	}	×
511
512	return 0;	1✔
513	}	1✔
514
515	int32_t azGetObjectsByPrefix(const char prefix, const char path) {	1✔
516	int32_t code = 0;	1✔
517
518	try {
519	code = azGetObjectsByPrefixImpl(prefix, path);	1!
520	} catch (const std::exception &e) {	×
521	azError("%s: Reason Phrase: %s", __func__, e.what());	×
522
523	code = TAOS_SYSTEM_ERROR(EIO);	×
524	azError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
525
526	TAOS_RETURN(code);	×
527	}	×
528
529	TAOS_RETURN(code);	1✔
530	}
531
532	int32_t azDeleteObjects(const char *object_name[], int nobject) {	1✔
533	for (int i = 0; i < nobject; ++i) {	2✔
534	azDeleteObjectsByPrefix(object_name[i]);	1✔
535	}
536
537	return 0;	1✔
538	}
539
540	#else
541
542	int32_t azCheckCfg() { return TSDB_CODE_SUCCESS; }
543
544	int32_t azPutObjectFromFileOffset(const char file, const char object_name, int64_t offset, int64_t size) {
545	return TSDB_CODE_SUCCESS;
546	}
547
548	int32_t azGetObjectBlock(const char object_name, int64_t offset, int64_t size, bool check, uint8_t *ppBlock) {
549	return TSDB_CODE_SUCCESS;
550	}
551
552	void azDeleteObjectsByPrefix(const char *prefix) {}
553
554	int32_t azPutObjectFromFile2(const char file, const char object, int8_t withcp) { return 0; }
555
556	int32_t azGetObjectsByPrefix(const char prefix, const char path) { return 0; }
557
558	int32_t azGetObjectToFile(const char object_name, const char fileName) { return 0; }
559
560	int32_t azDeleteObjects(const char *object_name[], int nobject) { return 0; }
561
562	#endif

taosdata / TDengine / #3842

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous