#3646

Committed 12 Mar 2025 12:34PM UTC coverage: 28.375% (-27.8%) from 56.156%

Build # #3646

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #30119 from taosdata/ciup30

ci: Update workflow to fix param issue of run_tdgpt_test

Run Details

59085 of 286935 branches covered (20.59%)

Branch coverage included in aggregate %.

102775 of 283490 relevant lines covered (36.25%)

55149.72 hits per line

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

0.0

/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; }	×
26
27	void azEnd() {}	×
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, ...) {	×
47	va_list arg_ptr;
48	va_start(arg_ptr, fmt);	×
49	(void)vfprintf(stderr, fmt, arg_ptr);	×
50	va_end(arg_ptr);	×
51	}	×
52
53	static void azDumpCfgByEp(int8_t epIndex) {	×
54	// clang-format off
55	checkPrint(	×
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],	×
63	"bucketName", tsS3BucketName,
64	"protocol", "https only",
65	"uristyle", "path only",
66	"accessKey", tsS3AccessKeyId[epIndex],	×
67	"accessKeySecret", tsS3AccessKeySecret[epIndex]);	×
68	// clang-format on
69	}	×
70
71	static int32_t azListBucket(char const *bucketname) {	×
72	int32_t code = 0;	×
73	const std::string delimiter = "/";	×
74	std::string accountName = tsS3AccessKeyId[0];	×
75	std::string accountKey = tsS3AccessKeySecret[0];	×
76	std::string accountURL = tsS3Hostname[0];	×
77	accountURL = "https://" + accountURL;	×
78
79	try {
80	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	×
81
82	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	×
83
84	std::string containerName = bucketname;	×
85	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	×
86
87	Azure::Storage::Blobs::ListBlobsOptions options;	×
88	options.Prefix = "s3";	×
89
90	checkPrint("objects:\n");	×
91	for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {	×
92	for (const auto &blob : pageResult.Blobs) {	×
93	checkPrint("%s\n", blob.Name.c_str());	×
94	}
95	}	×
96	} catch (const Azure::Core::RequestFailedException &e) {	×
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);	×
105	}	×
106
107	int32_t azCheckCfg() {	×
108	int32_t code = 0, lino = 0;	×
109
110	azDumpCfgByEp(0);	×
111
112	// test put
113	char testdata[17] = "0123456789abcdef";	×
114	const char *objectname[] = {"s3test.txt"};	×
115	char path[PATH_MAX] = {0};	×
116	int ds_len = strlen(TD_DIRSEP);	×
117	int tmp_len = strlen(tsTempDir);	×
118
119	(void)snprintf(path, PATH_MAX, "%s", tsTempDir);	×
120	if (strncmp(tsTempDir + tmp_len - ds_len, TD_DIRSEP, ds_len) != 0) {	×
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]);	×
125	}
126
127	uint8_t *pBlock = NULL;	×
128	int c_offset = 10;	×
129	int c_len = 6;	×
130	char buf[7] = {0};	×
131
132	TdFilePtr fp = taosOpenFile(path, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_READ \| TD_FILE_TRUNC);	×
133	if (!fp) {	×
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) {	×
138	checkPrint("failed to write test file: %s.\n", path);	×
139	TAOS_CHECK_GOTO(terrno, &lino, _next);	×
140	}
141	if (taosFsyncFile(fp) < 0) {	×
142	checkPrint("failed to fsync test file: %s.\n", path);	×
143	TAOS_CHECK_GOTO(terrno, &lino, _next);	×
144	}
145	(void)taosCloseFile(&fp);	×
146
147	checkPrint("\nstart to put object: %s, file: %s content: %s\n", objectname[0], path, testdata);	×
148	code = azPutObjectFromFileOffset(path, objectname[0], 0, 16);	×
149	if (code != 0) {	×
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]);	×
154
155	// list buckets
156	checkPrint("start to list bucket %s by prefix s3.\n", tsS3BucketName);	×
157	code = azListBucket(tsS3BucketName);	×
158	if (code != 0) {	×
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);	×
163
164	// test range get
165	checkPrint("start to range get object %s offset: %d len: %d.\n", objectname[0], c_offset, c_len);	×
166	code = azGetObjectBlock(objectname[0], c_offset, c_len, true, &pBlock);	×
167	if (code != 0) {	×
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);	×
173	taosMemoryFree(pBlock);	×
174	checkPrint("object content: %s\n", buf);	×
175	checkPrint("get object %s: success.\n\n", objectname[0]);	×
176
177	// delete test object
178	checkPrint("start to delete object: %s.\n", objectname[0]);	×
179	// code = azDeleteObjectsByPrefix(objectname[0]);
180	azDeleteObjectsByPrefix(objectname[0]);	×
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]);	×
188
189	_next:	×
190	if (fp) {	×
191	(void)taosCloseFile(&fp);	×
192	}
193
194	if (TSDB_CODE_SUCCESS != code) {	×
195	checkPrint("s3 check failed, code: %d, line: %d.\n", code, lino);	×
196	}
197
198	checkPrint("=================================================================\n");	×
199
200	TAOS_RETURN(code);	×
201	}
202
203	static int32_t azPutObjectFromFileOffsetImpl(const char file, const char object_name, int64_t offset, int64_t size) {	×
204	int32_t code = 0;	×
205
206	std::string endpointUrl = tsS3Hostname[0];	×
207	std::string accountName = tsS3AccessKeyId[0];	×
208	std::string accountKey = tsS3AccessKeySecret[0];	×
209
210	try {
211	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	×
212
213	std::string accountURL = tsS3Hostname[0];	×
214
215	accountURL = "https://" + accountURL;	×
216	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	×
217
218	std::string containerName = tsS3BucketName;	×
219	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	×
220
221	std::string blobName = "blob.txt";	×
222	uint8_t blobContent[] = "Hello Azure!";	×
223	// Create the block blob client
224	// BlockBlobClient blobClient = containerClient.GetBlockBlobClient(blobName);
225	TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));	×
226
227	blobClient.UploadFrom(file, offset, size);	×
228	} catch (const Azure::Core::RequestFailedException &e) {	×
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);	×
238	}	×
239
240	int32_t azPutObjectFromFileOffset(const char file, const char object_name, int64_t offset, int64_t size) {	×
241	int32_t code = 0;	×
242
243	try {
244	code = azPutObjectFromFileOffsetImpl(file, object_name, offset, size);	×
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);	×
255	}
256
257	static int32_t azGetObjectBlockImpl(const char *object_name, int64_t offset, int64_t size, bool check,	×
258	uint8_t **ppBlock) {
259	int32_t code = TSDB_CODE_SUCCESS;	×
260	std::string accountName = tsS3AccessKeyId[0];	×
261	std::string accountKey = tsS3AccessKeySecret[0];	×
262	std::string accountURL = tsS3Hostname[0];	×
263	uint8_t *buf = NULL;	×
264
265	try {
266	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	×
267
268	accountURL = "https://" + accountURL;	×
269	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	×
270
271	std::string containerName = tsS3BucketName;	×
272	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	×
273
274	TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));	×
275
276	Blobs::DownloadBlobToOptions options;	×
277	options.Range = Azure::Core::Http::HttpRange();	×
278	options.Range.Value().Offset = offset;	×
279	options.Range.Value().Length = size;	×
280
281	buf = (uint8_t *)taosMemoryCalloc(1, size);	×
282	if (!buf) {	×
283	return terrno;	×
284	}
285
286	auto res = blobClient.DownloadTo(buf, size, options);	×
287	if (check && res.Value.ContentRange.Length.Value() != size) {	×
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;	×
294	} catch (const Azure::Core::RequestFailedException &e) {	×
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);	×
309	}	×
310
311	static int32_t azGetObjectBlockRetry(const char *object_name, int64_t offset, int64_t size, bool check,	×
312	uint8_t **ppBlock) {
313	int32_t code = TSDB_CODE_SUCCESS;	×
314
315	// May use an exponential backoff policy for retries with 503
316	int retryCount = 0;	×
317	static int maxRetryCount = 5;
318	static int minRetryInterval = 1000; // ms
319	static int maxRetryInterval = 3000; // ms
320
321	_retry:	×
322	code = azGetObjectBlockImpl(object_name, offset, size, check, ppBlock);	×
323	if (TSDB_CODE_SUCCESS != code && retryCount++ < maxRetryCount) {	×
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);	×
330	}
331
332	int32_t azGetObjectBlock(const char object_name, int64_t offset, int64_t size, bool check, uint8_t *ppBlock) {	×
333	int32_t code = TSDB_CODE_SUCCESS;	×
334
335	try {
336	code = azGetObjectBlockRetry(object_name, offset, size, check, ppBlock);	×
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);	×
347	}
348
349	static void azDeleteObjectsByPrefixImpl(const char *prefix) {	×
350	const std::string delimiter = "/";	×
351	std::string accountName = tsS3AccessKeyId[0];	×
352	std::string accountKey = tsS3AccessKeySecret[0];	×
353	std::string accountURL = tsS3Hostname[0];	×
354	accountURL = "https://" + accountURL;	×
355
356	try {
357	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	×
358
359	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	×
360
361	std::string containerName = tsS3BucketName;	×
362	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	×
363
364	Azure::Storage::Blobs::ListBlobsOptions options;	×
365	options.Prefix = prefix;	×
366
367	std::set<std::string> listBlobs;	×
368	for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {	×
369	for (const auto &blob : pageResult.Blobs) {	×
370	listBlobs.insert(blob.Name);	×
371	}
372	}	×
373
374	for (auto blobName : listBlobs) {	×
375	auto blobClient = containerClient.GetAppendBlobClient(blobName);	×
376	blobClient.Delete();	×
377	}	×
378	} catch (const Azure::Core::RequestFailedException &e) {	×
379	azError("%s failed at line %d since %d(%s)", __func__, __LINE__, static_cast<int>(e.StatusCode),	×
380	e.ReasonPhrase.c_str());
381	}	×
382	}	×
383
384	void azDeleteObjectsByPrefix(const char *prefix) {	×
385	int32_t code = TSDB_CODE_SUCCESS;	×
386
387	try {
388	azDeleteObjectsByPrefixImpl(prefix);	×
389	} catch (const std::exception &e) {	×
390	azError("%s: Reason Phrase: %s", __func__, e.what());	×
391	}	×
392	}	×
393
394	int32_t azPutObjectFromFile2(const char file, const char object, int8_t withcp) {	×
395	int32_t code = 0, lino = 0;	×
396	uint64_t contentLength = 0;	×
397
398	if (taosStatFile(file, (int64_t *)&contentLength, NULL, NULL) < 0) {	×
399	azError("ERROR: %s Failed to stat file %s: ", __func__, file);	×
400	TAOS_RETURN(terrno);	×
401	}
402
403	code = azPutObjectFromFileOffset(file, object, 0, contentLength);	×
404	if (code != 0) {	×
405	azError("ERROR: %s Failed to put file %s: ", __func__, file);	×
406	TAOS_CHECK_GOTO(code, &lino, _exit);	×
407	}
408
409	_exit:	×
410	if (code) {	×
411	azError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
412	}
413
414	return 0;	×
415	}
416
417	static int32_t azGetObjectToFileImpl(const char object_name, const char fileName) {	×
418	int32_t code = TSDB_CODE_SUCCESS;	×
419	std::string accountName = tsS3AccessKeyId[0];	×
420	std::string accountKey = tsS3AccessKeySecret[0];	×
421	std::string accountURL = tsS3Hostname[0];	×
422	accountURL = "https://" + accountURL;	×
423
424	try {
425	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	×
426
427	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	×
428
429	std::string containerName = tsS3BucketName;	×
430	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	×
431
432	TDBlockBlobClient blobClient(containerClient.GetBlobClient(object_name));	×
433
434	auto res = blobClient.DownloadTo(fileName);	×
435	if (res.Value.ContentRange.Length.Value() <= 0) {	×
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) {	×
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);	×
449	}	×
450
451	int32_t azGetObjectToFile(const char object_name, const char fileName) {	×
452	int32_t code = 0;	×
453
454	try {
455	code = azGetObjectToFileImpl(object_name, fileName);	×
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);	×
466	}
467
468	static int32_t azGetObjectsByPrefixImpl(const char prefix, const char path) {	×
469	const std::string delimiter = "/";	×
470	std::string accountName = tsS3AccessKeyId[0];	×
471	std::string accountKey = tsS3AccessKeySecret[0];	×
472	std::string accountURL = tsS3Hostname[0];	×
473	accountURL = "https://" + accountURL;	×
474
475	try {
476	auto sharedKeyCredential = std::make_shared<StorageSharedKeyCredential>(accountName, accountKey);	×
477
478	BlobServiceClient blobServiceClient(accountURL, sharedKeyCredential);	×
479
480	std::string containerName = tsS3BucketName;	×
481	auto containerClient = blobServiceClient.GetBlobContainerClient(containerName);	×
482
483	Azure::Storage::Blobs::ListBlobsOptions options;	×
484	options.Prefix = prefix;	×
485
486	std::set<std::string> listBlobs;	×
487	for (auto pageResult = containerClient.ListBlobs(options); pageResult.HasPage(); pageResult.MoveToNextPage()) {	×
488	for (const auto &blob : pageResult.Blobs) {	×
489	listBlobs.insert(blob.Name);	×
490	}
491	}	×
492
493	for (auto blobName : listBlobs) {	×
494	const char *tmp = strchr(blobName.c_str(), '/');	×
495	tmp = (tmp == NULL) ? blobName.c_str() : tmp + 1;	×
496	char fileName[PATH_MAX] = {0};	×
497	if (path[strlen(path) - 1] != TD_DIRSEP_CHAR) {	×
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);	×
501	}
502	if (azGetObjectToFile(blobName.c_str(), fileName)) {	×
503	TAOS_RETURN(TSDB_CODE_FAILED);	×
504	}
505	}	×
506	} catch (const Azure::Core::RequestFailedException &e) {	×
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;	×
513	}	×
514
515	int32_t azGetObjectsByPrefix(const char prefix, const char path) {	×
516	int32_t code = 0;	×
517
518	try {
519	code = azGetObjectsByPrefixImpl(prefix, path);	×
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);	×
530	}
531
532	int32_t azDeleteObjects(const char *object_name[], int nobject) {	×
533	for (int i = 0; i < nobject; ++i) {	×
534	azDeleteObjectsByPrefix(object_name[i]);	×
535	}
536
537	return 0;	×
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 / #3646

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