enh: taos shell support cloud service 2.6 (#13880)

* enh: taos shell support cloud 2.6

* enh: taos shell support cloud (#13879)

* enh: support url in commandline taos shell

* fix: normal host logic

* enh: add default -R if env var set

* fix: refine taos shell cloud service

* fix: improve user friendly

* fix: add cloud tcp connection

* fix: header file

* fix: add other os

src/kit/shell/inc/shell.h

@@ -48,13 +48,11 @@ typedef struct SShellArguments {
   char* database;
   char* timezone;
   bool  restful;
-  char* token;
 #ifdef WINDOWS
   SOCKET socket;
 #else
   int socket;
 #endif
-
   TAOS* con;
   bool  is_raw_time;
   bool  is_use_passwd;
@@ -70,6 +68,11 @@ typedef struct SShellArguments {
   int   pktNum;
   char* pktType;
   char* netTestRole;
+  char* cloudDsn;
+  bool  cloud;
+  char* cloudHost;
+  char* cloudPort;
+  char* cloudToken;
 } SShellArguments;
 
 typedef enum WS_ACTION_TYPE_S { WS_CONN, WS_QUERY, WS_FETCH, WS_FETCH_BLOCK } WS_ACTION_TYPE;
@@ -91,7 +94,6 @@ void shellCheck(TAOS* con, SShellArguments* args);
 void get_history_path(char* history);
 void shellCheck(TAOS* con, SShellArguments* args);
 void cleanup_handler(void* arg);
-char *last_strstr(const char *haystack, const char *needle);
 void exitShell();
 int shellDumpResult(TAOS_RES* con, char* fname, int* error_no, bool printMode);
 void shellGetGrantInfo(void* con);
@@ -99,7 +101,8 @@ int isCommentLine(char* line);
 int wsclient_handshake();
 int wsclient_conn();
 void wsclient_query(char* command);
-int tcpConnect();
+int tcpConnect(char* host, int port);
+int parse_cloud_dsn();
 
 /**************** Global variable declarations ****************/
 extern char           PROMPT_HEADER[];

src/kit/shell/src/shellDarwin.c

@@ -62,8 +62,8 @@ void printHelp() {
   printf("%s%s%s\n", indent, indent, "Number of threads when using multi-thread to import data.");
   printf("%s%s\n", indent, "-R");
   printf("%s%s%s\n", indent, indent, "Connect and interact with TDengine use restful.");
-  printf("%s%s\n", indent, "-t");
-  printf("%s%s%s\n", indent, indent, "The token to use when connecting TDengine's cloud services.");
+  printf("%s%s\n", indent, "-E");
+  printf("%s%s%s\n", indent, indent, "The DSN to use when connecting TDengine's cloud services.");
   exit(EXIT_SUCCESS);
 }
 
@@ -76,20 +76,13 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
   for (int i = 1; i < argc; i++) {
     // for host
     if (strcmp(argv[i], "-h") == 0) {
-      if (i < argc - 1) {
-        char* arg = argv[++i];
-          char* tmp = strstr(arg, ":");
-          if (tmp == NULL) {
-              arguments->host = arg;
-          } else if ((tmp + 1) != NULL) {
-              arguments->port  = atoi(tmp + 1);
-              tmp[0] = '\0';
-              arguments->host = arg;
-          }
-      } else {
-        fprintf(stderr, "option -h requires an argument\n");
-        exit(EXIT_FAILURE);
-      }
+        if (i < argc - 1) {
+            arguments->cloud = false;
+            arguments->host = argv[++i];
+        } else {
+            fprintf(stderr, "option -h requires an argument\n");
+            exit(EXIT_FAILURE);
+        }
     }
       // for password
     else if ((strncmp(argv[i], "-p", 2) == 0)
@@ -116,6 +109,7 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
     // for management port
     else if (strcmp(argv[i], "-P") == 0) {
       if (i < argc - 1) {
+        arguments->cloud = false;
         arguments->port = atoi(argv[++i]);
       } else {
         fprintf(stderr, "option -P requires an argument\n");
@@ -132,6 +126,7 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
       }
     } else if (strcmp(argv[i], "-c") == 0) {
       if (i < argc - 1) {
+        arguments->cloud = false;
         if (strlen(argv[++i]) >= TSDB_FILENAME_LEN) {
           fprintf(stderr, "config file path: %s overflow max len %d\n", argv[i], TSDB_FILENAME_LEN - 1);
           exit(EXIT_FAILURE);
@@ -203,14 +198,15 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
     }
 
     else if (strcmp(argv[i], "-R") == 0) {
+        arguments->cloud = false;
         arguments->restful = true;
     }
 
-    else if (strcmp(argv[i], "-t") == 0) {
+    else if (strcmp(argv[i], "-E") == 0) {
         if (i < argc - 1) {
-            arguments->token = argv[++i];
+            arguments->cloudDsn = argv[++i];
         } else {
-            fprintf(stderr, "options -t requires an argument\n");
+            fprintf(stderr, "options -E requires an argument\n");
             exit(EXIT_FAILURE);
         }
     }
@@ -225,6 +221,16 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
       exit(EXIT_FAILURE);
     }
   }
+  if (args.cloudDsn == NULL) {
+      if (args.cloud) {
+          args.cloudDsn = getenv("TDENGINE_CLOUD_DSN");
+          if (args.cloudDsn == NULL) {
+              args.cloud = false;
+          }
+      }
+  } else {
+      args.cloud = true;
+  }
 }
 
 int32_t shellReadCommand(TAOS *con, char *command) {
@@ -573,23 +579,25 @@ void exitShell() {
   exit(EXIT_SUCCESS);
 }
 
-int tcpConnect() {
+int tcpConnect(char* host, int port) {
     struct sockaddr_in serv_addr;
-    if (args.port == 0) {
+    if (port == 0) {
+        port = 6041;
         args.port = 6041;
     }
-    if (NULL == args.host) {
+    if (NULL == host) {
+        host = "localhost";
         args.host = "localhost";
     }
 
-    struct hostent *server = gethostbyname(args.host);
+    struct hostent *server = gethostbyname(host);
     if ((server == NULL) || (server->h_addr == NULL)) {
-        fprintf(stderr, "no such host: %s\n", args.host);
+        fprintf(stderr, "no such host: %s\n", host);
         return -1;
     }
     memset(&serv_addr, 0, sizeof(struct sockaddr_in));
     serv_addr.sin_family = AF_INET;
-    serv_addr.sin_port = htons(args.port);
+    serv_addr.sin_port = htons(port);
     memcpy(&(serv_addr.sin_addr.s_addr), server->h_addr, server->h_length);
     args.socket = socket(AF_INET, SOCK_STREAM, 0);
     if (args.socket < 0) {

src/kit/shell/src/shellEngine.c

@@ -72,8 +72,7 @@ void shellInit(SShellArguments *_args) {
     _args->user = TSDB_DEFAULT_USER;
   }
 
-  if (_args->restful) {
-    _args->database = calloc(1, 128);
+  if (_args->restful || _args->cloud) {
     if (wsclient_handshake()) {
       exit(EXIT_FAILURE);
     }
@@ -159,7 +158,7 @@ static int32_t shellRunSingleCommand(TAOS *con, char *command) {
 
   // Analyse the command.
   if (regex_match(command, "^[ \t]*(quit|q|exit)[ \t;]*$", REG_EXTENDED | REG_ICASE)) {
-    if (args.restful) {
+    if (args.restful || args.cloud) {
       close(args.socket);
     } else {
       taos_close(con);
@@ -309,7 +308,7 @@ void shellRunCommandOnServer(TAOS *con, char command[]) {
     printMode = true;  // When output to a file, the switch does not work.
   }
 
-  if (args.restful) {
+  if (args.restful || args.cloud) {
     wsclient_query(command);
     return;
   }
@@ -1153,18 +1152,39 @@ int taos_base64_encode(unsigned char *source, size_t sourcelen, char *target, si
   return 1;
 }
 
-char *last_strstr(const char *haystack, const char *needle) {
-    if (*needle == '\0')
-        return (char *) haystack;
-
-    char *res = NULL;
-    for (;;) {
-        char *p = strstr(haystack, needle);
-        if (p == NULL) break;
-        res = p;
-        haystack = p + 1;
+int parse_cloud_dsn() {
+    if (args.cloudDsn == NULL) {
+        fprintf(stderr, "Cannot read cloud service info\n");
+        return 1;
+    } else {
+        char *start = strstr(args.cloudDsn, "http://");
+        if (start != NULL) {
+            args.cloudHost = start + strlen("http://");
+        } else {
+            start = strstr(args.cloudDsn, "https://");
+            if (start != NULL) {
+                args.cloudHost = start + strlen("https://");
+            } else {
+                args.cloudHost = args.cloudDsn;
+            }
+        }
+        char *port = strstr(args.cloudHost, ":");
+        if ((port == NULL) || (port + strlen(":")) == NULL) {
+            fprintf(stderr, "Invalid format in TDengine cloud dsn: %s\n", args.cloudDsn);
+            return 1;
+        }
+        char *token = strstr(port + strlen(":"), "?token=");
+        if ((token == NULL) || (token + strlen("?token=")) == NULL ||
+            (strlen(token + strlen("?token=")) == 0)) {
+            fprintf(stderr, "Invalid format in TDengine cloud dsn: %s\n", args.cloudDsn);
+            return -1;
+        }
+        port[0] = '\0';
+        args.cloudPort = port + strlen(":");
+        token[0] = '\0';
+        args.cloudToken = token + strlen("?token=");
     }
-    return res;
+    return 0;
 }
 
 int wsclient_handshake() {
@@ -1180,12 +1200,12 @@ int wsclient_handshake() {
     key_nonce[i] = rand() & 0xff;
   }
   taos_base64_encode(key_nonce, 16, websocket_key, 256);
-  if (args.token) {
-    snprintf(request_header, 1024,
-             "GET /rest/ws?token=%s HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nHost: "
-             "%s:%d\r\nSec-WebSocket-Key: "
-             "%s\r\nSec-WebSocket-Version: 13\r\n\r\n",
-             args.token, args.host, args.port, websocket_key);
+  if (args.cloud) {
+        snprintf(request_header, 1024,
+                 "GET /rest/ws?token=%s HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nHost: "
+                 "%s:%s\r\nSec-WebSocket-Key: "
+                 "%s\r\nSec-WebSocket-Version: 13\r\n\r\n",
+                args.cloudToken, args.cloudHost, args.cloudPort, websocket_key);
   } else {
     snprintf(request_header, 1024,
              "GET /rest/ws HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nHost: %s:%d\r\nSec-WebSocket-Key: "
@@ -1286,9 +1306,9 @@ int wsclient_send_sql(char *command, WS_ACTION_TYPE type, int id) {
   switch (type) {
     case WS_CONN:
       cJSON_AddStringToObject(json, "action", "conn");
-      cJSON_AddStringToObject(_args, "user", "root");
-      cJSON_AddStringToObject(_args, "password", "taosdata");
-      cJSON_AddStringToObject(_args, "db", "");
+      cJSON_AddStringToObject(_args, "user", args.user);
+      cJSON_AddStringToObject(_args, "password", args.password);
+      cJSON_AddStringToObject(_args, "db", args.database);
 
       break;
     case WS_QUERY:
@@ -1341,6 +1361,12 @@ int wsclient_conn() {
   }
   if (code->valueint == 0) {
     cJSON_Delete(root);
+    if (args.cloud) {
+        fprintf(stdout, "Successfully connect to %s:%s in restful mode\n\n", args.cloudHost, args.cloudPort);
+    } else {
+        fprintf(stdout, "Successfully connect to %s:%d in restful mode\n\n", args.host, args.port);
+    }
+
     return 0;
   } else {
     cJSON *message = cJSON_GetObjectItem(root, "message");
@@ -1606,4 +1632,4 @@ void wsclient_query(char *command) {
   }
   cJSON_Delete(query);
   return;
-}
+}
\ No newline at end of file

src/kit/shell/src/shellLinux.c

@@ -54,7 +54,7 @@ static struct argp_option options[] = {
   {"pktnum",     'N', "PKTNUM",     0,                   "Packet numbers used for net test, default is 100."},
   {"pkttype",    'S', "PKTTYPE",    0,                   "Choose packet type used for net test, default is TCP. Only speed test could be either TCP or UDP."},
   {"restful", 'R', 0, 0, "Connect and interact with TDengine use restful."},
-  {"token", 't', "TOKEN", 0, "The token to use when connecting TDengine's cloud services."},
+  {0, 'E', "DSN", 0, "The DSN to use when connecting TDengine's cloud services."},
   {0}};
 
 static error_t parse_opt(int key, char *arg, struct argp_state *state) {
@@ -64,21 +64,20 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
   wordexp_t full_path;
 
   switch (key) {
-    case 'h':{
-      char* tmp = strstr(arg, ":");
-      if (tmp == NULL) {
-        arguments->host = arg;
-      } else if ((tmp + 1) != NULL) {
-        arguments->port  = atoi(tmp + 1);
-        tmp[0] = '\0';
-        arguments->host = arg;
-      }
-      break;
-    }
+    case 'h':
+        if (arg) {
+          args.cloud = false;
+          args.host = arg;
+        } else {
+          fprintf(stderr, "Invalid host\n");
+          return -1;
+        }
+        break;
     case 'p':
       break;
     case 'P':
       if (arg) {
+        args.cloud = false;
         tsDnodeShellPort = atoi(arg);
         p_port = atoi(arg);
       } else {
@@ -106,6 +105,7 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
         wordfree(&full_path);
         return -1;
       }
+      args.cloud = false;
       tstrncpy(configDir, full_path.we_wordv[0], TSDB_FILENAME_LEN);
       wordfree(&full_path);
       break;
@@ -173,11 +173,17 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
     case OPT_ABORT:
       arguments->abort = 1;
       break;
-    case 't':
-      arguments->token = arg;
-      break;
     case 'R':
       arguments->restful = true;
+      arguments->cloud = false;
+      break;
+    case 'E':
+      if (arg) {
+         arguments->cloudDsn = arg;
+       } else {
+         fprintf(stderr, "Invalid -E option\n");
+         return -1;
+       }
       break;
     default:
       return ARGP_ERR_UNKNOWN;
@@ -231,8 +237,16 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
   }
 
   argp_parse(&argp, argc, argv, 0, 0, arguments);
-  if (arguments->token == NULL) {
-    arguments->port = p_port;
+
+  if (args.cloudDsn == NULL) {
+      if (args.cloud) {
+          args.cloudDsn = getenv("TDENGINE_CLOUD_DSN");
+          if (args.cloudDsn == NULL) {
+              args.cloud = false;
+          }
+      }
+  } else {
+      args.cloud = true;
   }
   
   if (arguments->abort) {
@@ -595,23 +609,25 @@ void exitShell() {
   exit(EXIT_SUCCESS);
 }
 
-int tcpConnect() {
+int tcpConnect(char* host, int port) {
     struct sockaddr_in serv_addr;
-    if (args.port == 0) {
+    if (port == 0) {
+        port = 6041;
         args.port = 6041;
     }
-    if (NULL == args.host) {
+    if (NULL == host) {
+        host = "localhost";
         args.host = "localhost";
     }
 
-    struct hostent *server = gethostbyname(args.host);
+    struct hostent *server = gethostbyname(host);
     if ((server == NULL) || (server->h_addr == NULL)) {
-        fprintf(stderr, "no such host: %s\n", args.host);
+        fprintf(stderr, "no such host: %s\n", host);
         return -1;
     }
     memset(&serv_addr, 0, sizeof(struct sockaddr_in));
     serv_addr.sin_family = AF_INET;
-    serv_addr.sin_port = htons(args.port);
+    serv_addr.sin_port = htons(port);
     memcpy(&(serv_addr.sin_addr.s_addr), server->h_addr, server->h_length);
     args.socket = socket(AF_INET, SOCK_STREAM, 0);
     if (args.socket < 0) {

src/kit/shell/src/shellMain.c

@@ -76,7 +76,6 @@ SShellArguments args = {.host = NULL,
   .database = NULL,
   .timezone = NULL,
   .restful = false,
-  .token = NULL,
   .is_raw_time = false,
   .is_use_passwd = false,
   .dump_config = false,
@@ -87,7 +86,12 @@ SShellArguments args = {.host = NULL,
   .pktLen = 1000,
   .pktNum = 100,
   .pktType = "TCP",
-  .netTestRole = NULL};
+  .netTestRole = NULL,
+  .cloud = true,
+  .cloudHost = NULL,
+  .cloudPort = NULL,
+  .cloudToken = NULL,
+  };
 
 /*
  * Main function.
@@ -102,35 +106,6 @@ int main(int argc, char* argv[]) {
     exit(EXIT_FAILURE);
   }
 
-  char* cloud_url = getenv("TDENGINE_CLOUD_URL");
-  if (cloud_url != NULL) {
-    char* start = strstr(cloud_url, "http://");
-    if (start != NULL) {
-      cloud_url = start + strlen("http://");
-    } else {
-      start = strstr(cloud_url, "https://");
-      if (start != NULL) {
-        cloud_url = start + strlen("https://");
-      }
-    }
-    
-    char* tmp = last_strstr(cloud_url, ":");
-    if ((tmp == NULL) && ((tmp + 1) != NULL )) {
-      fprintf(stderr, "Invalid format in environment variable TDENGINE_CLOUD_URL: %s\n", cloud_url);
-      exit(EXIT_FAILURE);
-    } else {
-      args.port = atoi(tmp + 1);
-      tmp[0] = '\0';
-      args.host = cloud_url;
-    }
-  }
-
-  char* cloud_token = getenv("TDENGINE_CLOUD_TOKEN");
-  
-  if (cloud_token != NULL) {
-    args.token = cloud_token;
-  }
-
   shellParseArgument(argc, argv, &args);
 
   if (args.dump_config) {
@@ -155,10 +130,17 @@ int main(int argc, char* argv[]) {
     exit(0);
   }
 
-  if (args.restful) {
-    if (tcpConnect()) {
-      exit(EXIT_FAILURE);
-    }
+  if (args.cloud) {
+      if (parse_cloud_dsn()) {
+          exit(EXIT_FAILURE);
+      }
+      if (tcpConnect(args.cloudHost, atoi(args.cloudPort))) {
+          exit(EXIT_FAILURE);
+      }
+  } else if (args.restful) {
+      if (tcpConnect(args.host, args.port)) {
+          exit(EXIT_FAILURE);
+      }
   }
 
   /* Initialize the shell */

src/kit/shell/src/shellWindows.c

@@ -64,8 +64,8 @@ void printHelp() {
   printf("%s%s%s\n", indent, indent, "Packet numbers used for net test, default is 100.");
   printf("%s%s\n", indent, "-R");
   printf("%s%s%s\n", indent, indent, "Connect and interact with TDengine use restful.");
-  printf("%s%s\n", indent, "-t");
-  printf("%s%s%s\n", indent, indent, "The token to use when connecting TDengine's cloud services.");
+  printf("%s%s\n", indent, "-E");
+  printf("%s%s%s\n", indent, indent, "The DSN to use when connecting TDengine's cloud services.");
   printf("%s%s\n", indent, "-S");
   printf("%s%s%s\n", indent, indent, "Packet type used for net test, default is TCP.");
   printf("%s%s\n", indent, "-V");
@@ -80,15 +80,8 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
     // for host
     if (strcmp(argv[i], "-h") == 0) {
       if (i < argc - 1) {
-        char* arg = argv[++i];
-          char* tmp = strstr(arg, ":");
-          if (tmp == NULL) {
-              arguments->host = arg;
-          } else if ((tmp + 1) != NULL) {
-              arguments->port  = atoi(tmp + 1);
-              tmp[0] = '\0';
-              arguments->host = arg;
-          }
+          arguments->cloud = false;
+          arguments->host = argv[++i];
       } else {
         fprintf(stderr, "option -h requires an argument\n");
         exit(EXIT_FAILURE);
@@ -119,6 +112,7 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
     // for management port
     else if (strcmp(argv[i], "-P") == 0) {
       if (i < argc - 1) {
+        arguments->cloud = false;
         arguments->port = atoi(argv[++i]);
       } else {
         fprintf(stderr, "option -P requires an argument\n");
@@ -142,6 +136,7 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
       }
     } else if (strcmp(argv[i], "-c") == 0) {
       if (i < argc - 1) {
+        arguments->cloud = false;
         char *tmp = argv[++i];
         if (strlen(tmp) >= TSDB_FILENAME_LEN) {
           fprintf(stderr, "config file path: %s overflow max len %d\n", tmp, TSDB_FILENAME_LEN - 1);
@@ -225,14 +220,15 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
     }
 
     else if (strcmp(argv[i], "-R") == 0) {
+        arguments->cloud = false;
         arguments->restful = true;
     }
 
-    else if (strcmp(argv[i], "-t") == 0) {
+    else if (strcmp(argv[i], "-E") == 0) {
         if (i < argc - 1) {
-            arguments->token = argv[++i];
+            arguments->cloudDsn = argv[++i];
         } else {
-            fprintf(stderr, "options -t requires an argument\n");
+            fprintf(stderr, "options -E requires an argument\n");
             exit(EXIT_FAILURE);
         }
     }
@@ -251,6 +247,16 @@ void shellParseArgument(int argc, char *argv[], SShellArguments *arguments) {
       exit(EXIT_FAILURE);
     }
   }
+  if (args.cloudDsn == NULL) {
+      if (args.cloud) {
+          args.cloudDsn = getenv("TDENGINE_CLOUD_DSN");
+          if (args.cloudDsn == NULL) {
+              args.cloud = false;
+          }
+      }
+  } else {
+      args.cloud = true;
+  }
 }
 
 void shellPrintContinuePrompt() { printf("%s", CONTINUE_PROMPT); }
@@ -363,21 +369,23 @@ void get_history_path(char *history) {
 
 void exitShell() { exit(EXIT_SUCCESS); }
 
-int tcpConnect() {
+int tcpConnect(char* host, int iport) {
     int iResult;
     WSADATA wsaData;
     struct addrinfo *aResult = NULL,
             *ptr = NULL,
             hints;
-    if (args.port == 0) {
-        args.port = 6041;
+    if (iport == 0) {
+        iport = 6041;
+        args.port = iport;
     }
-    if (NULL == args.host) {
+    if (NULL == host) {
+        host = "localhost";
         args.host = "localhost";
     }
     char port[10] = {0};
 
-    sprintf_s(port, 10, "%d", args.port);
+    sprintf_s(port, 10, "%d", iport);
 
     iResult = WSAStartup(MAKEWORD(2,2), &wsaData);
     if (iResult != 0) {
@@ -388,7 +396,7 @@ int tcpConnect() {
     hints.ai_family = AF_UNSPEC;
     hints.ai_socktype = SOCK_STREAM;
     hints.ai_protocol = IPPROTO_TCP;
-    iResult = getaddrinfo(args.host, port, &hints, &aResult);
+    iResult = getaddrinfo(host, port, &hints, &aResult);
     if ( iResult != 0 ) {
         printf("getaddrinfo failed with error: %d\n", iResult);
         WSACleanup();