Merge pull request #19808 from taosdata/enh/stream_num_limit

enh: stream number limit

include/util/taoserror.h

@@ -355,6 +355,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_STREAM_MUST_BE_DELETED    TAOS_DEF_ERROR_CODE(0, 0x03F3)
 #define TSDB_CODE_MND_STREAM_TASK_DROPPED       TAOS_DEF_ERROR_CODE(0, 0x03F4)
 #define TSDB_CODE_MND_MULTI_REPLICA_SOURCE_DB   TAOS_DEF_ERROR_CODE(0, 0x03F5)
+#define TSDB_CODE_MND_TOO_MANY_STREAMS          TAOS_DEF_ERROR_CODE(0, 0x03F6)
 
 // mnode-sma
 #define TSDB_CODE_MND_SMA_ALREADY_EXIST         TAOS_DEF_ERROR_CODE(0, 0x0480)

source/dnode/mnode/impl/src/mndStream.c

@@ -31,6 +31,8 @@
 #define MND_STREAM_VER_NUMBER   2
 #define MND_STREAM_RESERVE_SIZE 64
 
+#define MND_STREAM_MAX_NUM 10
+
 static int32_t mndStreamActionInsert(SSdb *pSdb, SStreamObj *pStream);
 static int32_t mndStreamActionDelete(SSdb *pSdb, SStreamObj *pStream);
 static int32_t mndStreamActionUpdate(SSdb *pSdb, SStreamObj *pStream, SStreamObj *pNewStream);
@@ -625,6 +627,35 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  {
+    int32_t numOfStream = 0;
+
+    SStreamObj *pStream = NULL;
+    void       *pIter = NULL;
+
+    while (1) {
+      pIter = sdbFetch(pMnode->pSdb, SDB_STREAM, pIter, (void **)&pStream);
+      if (pIter == NULL) {
+        if (numOfStream > MND_STREAM_MAX_NUM) {
+          mError("too many streams, no more than 10 for each database");
+          terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
+          goto _OVER;
+        }
+        break;
+      }
+
+      if (pStream->sourceDbUid == streamObj.sourceDbUid) {
+        ++numOfStream;
+      }
+      sdbRelease(pMnode->pSdb, pStream);
+      if (numOfStream > MND_STREAM_MAX_NUM) {
+        mError("too many streams, no more than 10 for each database");
+        terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
+        goto _OVER;
+      }
+    }
+  }
+
   pDb = mndAcquireDb(pMnode, streamObj.sourceDb);
   if (pDb->cfg.replications != 1) {
     mError("stream source db must have only 1 replica, but %s has %d", pDb->name, pDb->cfg.replications);

source/util/src/terror.c

@@ -287,6 +287,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_STREAM_NOT_EXIST,         "Stream not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STREAM_OPTION,    "Invalid stream option")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_STREAM_MUST_BE_DELETED,   "Stream must be dropped first")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_MULTI_REPLICA_SOURCE_DB,  "Stream temporarily does not support source db having replica > 1")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_STREAMS,         "Too many streams")
 
 // mnode-sma
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_SMA_ALREADY_EXIST,        "SMA already exists")