feat(streaming): onboard streaming module (#1024)

Signed-off-by: github-actions <github-actions@github.com>
Co-authored-by: github-actions <github-actions@github.com>
Co-authored-by: Meng Xin Zhu <843303+zxkane@users.noreply.github.com>

.projenrc.js

@@ -90,6 +90,8 @@ const awsSDKServicesDepsForApiProject = [
   '@aws-sdk/client-s3',
   '@aws-sdk/client-sfn',
   '@aws-sdk/client-kafka',
+  '@aws-sdk/client-kms',
+  '@aws-sdk/client-kinesis-analytics-v2',
   '@aws-sdk/client-redshift',
   '@aws-sdk/client-redshift-data',
   '@aws-sdk/client-redshift-serverless',

.viperlightignore

@@ -43,7 +43,6 @@ src/base-lib/src/constant/constant.ts:80
 src/base-lib/src/constant/constant.ts:83
 src/base-lib/src/constant/constant.ts:86
 src/base-lib/src/constant/constant.ts:89
-src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:60
 src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:61
 src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:62
 src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:63
@@ -53,6 +52,7 @@ src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:66
 src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:67
 src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:68
 src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:69
+src/control-plane/backend/lambda/api/test/api/ddb-mock.ts:70
 src/control-plane/backend/lambda/api/common/constants.ts:14
 
 [node-yarnoutdated]

NOTICE.txt

@@ -31,6 +31,8 @@ This software includes third party software subject to the following copyrights:
 @aws-sdk/client-iam under the Apache-2.0
 @aws-sdk/client-kafka under the Apache-2.0
 @aws-sdk/client-kafkaconnect under the Apache-2.0
+@aws-sdk/client-kinesis-analytics-v2 under the Apache-2.0
+@aws-sdk/client-kms under the Apache-2.0
 @aws-sdk/client-lambda under the Apache-2.0
 @aws-sdk/client-quicksight under the Apache-2.0
 @aws-sdk/client-redshift under the Apache-2.0

frontend/public/locales/en-US/application.json

@@ -69,5 +69,9 @@
     "pipelineInfo": "Pipeline Information",
     "serverEdp": "Server Endpoint",
     "serverDomain": "Server Domain"
+  },
+  "delete": {
+    "title": "Delete Application",
+    "tip": "Are you sure you want to delete the application {{appId}} ?"
   }
 }

frontend/public/locales/en-US/common.json

@@ -216,7 +216,9 @@
     "apply": "Apply",
     "newGroup": "Add Group",
     "export": "Export",
-    "refreshSegment": "Refresh Segment"
+    "refreshSegment": "Refresh Segment",
+    "streamEnable": "Streaming Enable",
+    "streamDisable": "Streaming Disable"
   },
   "item": "item",
   "items": "items",

frontend/public/locales/en-US/pipeline.json

@@ -312,7 +312,13 @@
     "redshiftProvisionedDesc": "Provisioned Redshift cluster that meet your cost and performance specifications",
     "reportNotSupported": "Not Supported",
     "reportNotSupportedDesc": "Reporting is not supported if data processing is not enabled.",
-    "notSupportedServices": "These service(s): {{unSupportedServices}} are not available in this region, some of features may not supported."
+    "notSupportedServices": "These service(s): {{unSupportedServices}} are not available in this region, some of features may not supported.",
+    "streamingTitle": "Real-time analysis settings",
+    "streamingDesc": "Enable real-time analysis for this pipeline",
+    "enableStreaming": "Enable real-time analysis",
+    "enableStreamingDesc": "Enabling real-time analysis will create the real-time data processing resources (i.e., Lambda) for this pipeline. Note that additional cost apply.",
+    "streamingDataRangeTitle": "Real-time data range",
+    "streamingDataRangeDesc": "Real-time data range is the period to keep the real-time data for, data exceed the data range will be deleted from the real-time data table and won’t be displayed in the dashboard."
   },
   "detail": {
     "ingestion": "Ingestion",

frontend/public/locales/en-US/project.json

@@ -52,7 +52,8 @@
     "noAppDisplay": "No application to display.",
     "findApp": "Find applications",
     "yourApp": "Your applications",
-    "yourAppDesc": "This table lists all the applications registered in this project"
+    "yourAppDesc": "This table lists all the applications registered in this project",
+    "appStreamEnable": "Streaming enable"
   },
   "list": {
     "name": "Project Name",
@@ -70,4 +71,4 @@
     "tip1": "Are you sure you want to delete the project ",
     "tip2": " delete project will delete all data in redshift."
   }
-}
+}

frontend/public/locales/zh-CN/application.json

@@ -69,5 +69,9 @@
     "pipelineInfo": "数据管道信息",
     "serverEdp": "服务器端点",
     "serverDomain": "服务器域"
+  },
+  "delete": {
+    "title": "删除应用程序",
+    "tip": "您确定要删除该应用程序  {{appId}} 吗？"
   }
 }

frontend/public/locales/zh-CN/common.json

@@ -216,7 +216,9 @@
     "apply": "应用",
     "newGroup": "添加分组",
     "export": "导出",
-    "refreshSegment": "刷新分群"
+    "refreshSegment": "刷新分群",
+    "streamEnable": "开启实时流",
+    "streamDisable": "关闭实时流"
   },
   "item": "项目",
   "items": "项目",

frontend/public/locales/zh-CN/pipeline.json

@@ -312,7 +312,13 @@
     "redshiftProvisionedDesc": "选择符合您的成本和性能规格的预置 Redshift 集群",
     "reportNotSupported": "不支持",
     "reportNotSupportedDesc": "如果未启用数据处理，则不支持自动生成分析报表。",
-    "notSupportedServices": "以下服务：{{unSupportedServices}} 在此区域不可用，某些功能可能不支持。"
+    "notSupportedServices": "以下服务：{{unSupportedServices}} 在此区域不可用，某些功能可能不支持。",
+    "streamingTitle": "实时分析配置",
+    "streamingDesc": "启用此管道的实时分析",
+    "enableStreaming": "启用实时分析",
+    "enableStreamingDesc": "启用实时分析将为该管道创建实时数据处理资源（即Lambda）。请注意，需要支付额外费用。",
+    "streamingDataRangeTitle": "实时数据范围",
+    "streamingDataRangeDesc": "实时数据范围是保留实时数据的时间段，超过数据范围的数据将从实时数据表中删除，不会显示在仪表板中。"
   },
   "detail": {
     "ingestion": "数据摄取",

frontend/public/locales/zh-CN/project.json

@@ -52,7 +52,8 @@
     "noAppDisplay": "没有可显示的应用程序。",
     "findApp": "查找应用程序",
     "yourApp": "你的应用程序",
-    "yourAppDesc": "此表列出了在此项目中注册的所有应用程序"
+    "yourAppDesc": "此表列出了在此项目中注册的所有应用程序",
+    "appStreamEnable": "启用实时流"
   },
   "list": {
     "name": "项目名称",

frontend/src/apis/application.ts

@@ -55,10 +55,20 @@ const updateApplicationTimezone = async (params: {
   return result;
 };
 
+const streamApplication = async (data: {
+  id: string;
+  pid: string;
+  enable: boolean;
+}) => {
+  const result: any = await apiRequest('put', `/app/${data.id}/stream`, data);
+  return result;
+};
+
 export {
   getApplicationListByPipeline,
   createApplication,
   getApplicationDetail,
   deleteApplication,
   updateApplicationTimezone,
+  streamApplication,
 };

frontend/src/pages/pipelines/comps/BasicInfo.tsx

@@ -233,7 +233,8 @@ const BasicInfo: React.FC<BasicInfoProps> = (props: BasicInfoProps) => {
                     {t('button.edit')}
                   </Button>
                 )}
-                {pipelineInfo?.statusType === EPipelineStatus.Active && (
+                {(pipelineInfo?.statusType === EPipelineStatus.Active ||
+                  pipelineInfo?.statusType === EPipelineStatus.Warning) && (
                   <Button
                     iconName="upload-download"
                     disabled={

frontend/src/pages/pipelines/create/CreatePipeline.tsx

@@ -920,6 +920,15 @@ const Content: React.FC<ContentProps> = (props: ContentProps) => {
         createPipelineObj.dataModeling.redshift.provisioned = null;
       }
     }
+
+    // set streaming
+    if (!pipelineInfo.enableStreaming || !pipelineInfo.enableRedshift) {
+      createPipelineObj.streaming = null;
+    } else {
+      createPipelineObj.streaming = {
+        appIdStreamList: pipelineInfo.streaming?.appIdStreamList ?? [],
+      };
+    }
     return createPipelineObj;
   };
 
@@ -995,6 +1004,7 @@ const Content: React.FC<ContentProps> = (props: ContentProps) => {
       'kafkaBrokers',
       'arnAccountId',
       'enableReporting',
+      'enableStreaming',
       'selectedQuickSightUser',
       'dataConnectionType',
       'quickSightVpcConnection',
@@ -1296,6 +1306,14 @@ const Content: React.FC<ContentProps> = (props: ContentProps) => {
               kafkaSGEmptyError={kafkaSGEmptyError}
               bufferKDSModeEmptyError={bufferKDSModeEmptyError}
               bufferKDSShardNumFormatError={bufferKDSShardNumFormatError}
+              changeEnableStreaming={(enable) => {
+                setPipelineInfo((prev) => {
+                  return {
+                    ...prev,
+                    enableStreaming: enable,
+                  };
+                });
+              }}
               changeNetworkType={(type) => {
                 setPipelineInfo((prev) => {
                   return {
@@ -1536,6 +1554,7 @@ const Content: React.FC<ContentProps> = (props: ContentProps) => {
               changeBufferType={(type) => {
                 let sinkInterval = '';
                 let sinkBatchSize = '';
+                let tmpEnableStream = false;
                 if (type === SinkType.KDS) {
                   sinkInterval = DEFAULT_KDS_SINK_INTERVAL;
                   sinkBatchSize = DEFAULT_KDS_BATCH_SIZE;
@@ -1567,11 +1586,16 @@ const Content: React.FC<ContentProps> = (props: ContentProps) => {
                   }
                 }
 
+                if (type === SinkType.S3 || type === SinkType.MSK) {
+                  tmpEnableStream = false;
+                }
+
                 setPipelineInfo((prev) => {
                   return {
                     ...prev,
                     enableDataProcessing: tmpEnableProcessing,
                     enableReporting: tmpEnableQuickSight,
+                    enableStreaming: tmpEnableStream,
                     ingestionServer: {
                       ...prev.ingestionServer,
                       sinkType: type,
@@ -1819,6 +1843,14 @@ const Content: React.FC<ContentProps> = (props: ContentProps) => {
                   };
                 });
               }}
+              changeStreamingDataRangeValue={(value) => {
+                setPipelineInfo((prev) => {
+                  return {
+                    ...prev,
+                    streamingDataRangeValue: value,
+                  };
+                });
+              }}
             />
           ),
         },
@@ -2635,6 +2667,11 @@ const CreatePipeline: React.FC<CreatePipelineProps> = (
     )[0];
   };
 
+  const setUpdateStreaming = async (pipelineInfo: IExtPipeline) => {
+    pipelineInfo.enableStreaming =
+      pipelineInfo.streaming?.appIdStreamList !== undefined;
+  };
+
   const setUpdateQuickSightUser = async (pipelineInfo: IExtPipeline) => {
     if (!pipelineInfo.reporting?.quickSight.user) {
       return;
@@ -2850,6 +2887,8 @@ const CreatePipeline: React.FC<CreatePipelineProps> = (
           newServerless: data.dataModeling?.redshift?.newServerless ?? null,
         },
       },
+      streaming: data.streaming,
+      reporting: data.reporting,
       statusType: data.statusType,
       stackDetails: defaultGenericsValue(data.stackDetails, []),
       executionDetail: data.executionDetail,
@@ -2894,6 +2933,7 @@ const CreatePipeline: React.FC<CreatePipelineProps> = (
               : setUpdateMSKCluster(extPipeline),
             setUpdateKDSType(extPipeline),
             setUpdateETL(extPipeline),
+            setUpdateStreaming(extPipeline),
             setUpdateReport(extPipeline),
           ])
             .then(() => {