feat(ingest): support log ingestion with pipeline (#521)

* feat: init ingest

* refactor(ingest): components rewrite

* refactor: rename

* feat: remove components

* feat(ingest): log ingestion with pipeline

* style: new icon for log ingestion

* feat(ingest): select format for log ingestion

* docs: format

* style: sidebar width

* feat(ingest): responsive codemirror format and placeholder

Author: ZonaHex

.vscode/settings.json

@@ -7,6 +7,7 @@
     "Greptime",
     "hoverable",
     "Keymap",
+    "ndjson",
     "pinia",
     "promql",
     "resizebox",

src/api/pipeline.ts

@@ -6,6 +6,7 @@ import editorAPI from './editor'
 const { runSQL } = editorAPI
 const url = '/v1/events/pipelines'
 const sqlUrl = `/v1/sql`
+const pipelineLogsUrl = '/v1/events/logs'
 
 const makeSqlData = (sql: string) => {
   return qs.stringify({
@@ -58,6 +59,25 @@ export function list() {
   })
 }
 
+export const postPipelineLogs = (
+  db: string,
+  table: string,
+  pipelineName: string,
+  data: string,
+  contentType = 'application/x-ndjson'
+) => {
+  return axios.post(pipelineLogsUrl, data, {
+    params: {
+      db,
+      table,
+      pipeline_name: pipelineName,
+    },
+    headers: {
+      'Content-Type': contentType,
+    },
+  })
+}
+
 export function getByName(name: string): Promise<PipeFile> {
   const sql = `select name, created_at, pipeline 
     from greptime_private.pipelines 

src/assets/icons.js

@@ -14,6 +14,24 @@ export interface IngestConfig {
 export default function useIngest() {
   const codeRunStore = useCodeRunStore()
 
+  const getPlaceholderByContentType = (contentType: string): string => {
+    if (contentType === 'application/json') {
+      return `[
+  {"message": "hello world", "time": "2024-07-12T16:18:53.048"},
+  {"message": "hello greptime", "time": "2024-07-12T16:18:53.048"}
+]`
+    }
+
+    if (contentType === 'application/x-ndjson') {
+      return `{"message": "hello world", "time": "2024-07-12T16:18:53.048"}
+{"message": "hello greptime", "time": "2024-07-12T16:18:53.048"}`
+    }
+
+    // text/plain
+    return `hello world 2024-07-12T16:18:53.048
+hello greptime 2024-07-12T16:18:53.048`
+  }
+
   const getInfluxdbInputConfig = (precision: Ref<string>): IngestConfig => ({
     type: 'influxdb',
     tabKey: 'influxdb-input',
@@ -29,19 +47,6 @@ export default function useIngest() {
     },
   })
 
-  const getLogIngestionInputConfig = (pipeline: Ref<string>): IngestConfig => ({
-    type: 'log-ingestion',
-    tabKey: 'log-ingestion-input',
-    submitLabel: 'Process',
-    placeholder: 'Enter your log data here...',
-    hasDoc: false,
-    successMessage: 'Log processed successfully',
-    submitHandler: async (content: string) => {},
-    get params() {
-      return { pipeline: pipeline.value }
-    },
-  })
-
   const getInfluxdbUploadConfig = (precision: Ref<string>): IngestConfig => ({
     type: 'influxdb',
     tabKey: 'influxdb-upload',
@@ -56,15 +61,45 @@ export default function useIngest() {
     },
   })
 
-  const getLogIngestionUploadConfig = (pipeline: Ref<string>): IngestConfig => ({
+  const getLogIngestionInputConfig = (
+    pipeline: Ref<string>,
+    table: Ref<string>,
+    contentType: Ref<string>
+  ): IngestConfig => ({
+    type: 'log-ingestion',
+    tabKey: 'log-ingestion-input',
+    submitLabel: 'Write',
+    get placeholder() {
+      return getPlaceholderByContentType(contentType.value)
+    },
+    hasDoc: false,
+    successMessage: 'Data written successfully',
+    submitHandler: async (content: string) => {
+      return codeRunStore.processLogs(content, table.value, pipeline.value, contentType.value)
+    },
+    get params() {
+      return { pipeline: pipeline.value, table: table.value, contentType: contentType.value }
+    },
+  })
+
+  const getLogIngestionUploadConfig = (
+    pipeline: Ref<string>,
+    table: Ref<string>,
+    contentType: Ref<string>
+  ): IngestConfig => ({
     type: 'log-ingestion',
     tabKey: 'log-ingestion-upload',
-    submitLabel: 'Process',
+    submitLabel: 'Write',
+    get placeholder() {
+      return getPlaceholderByContentType(contentType.value)
+    },
     hasDoc: false,
-    successMessage: 'Log processed successfully',
-    submitHandler: async (content: string) => {},
+    successMessage: 'Data written successfully',
+    submitHandler: async (content: string) => {
+      return codeRunStore.processLogs(content, table.value, pipeline.value, contentType.value)
+    },
     get params() {
-      return { pipeline: pipeline.value }
+      return { pipeline: pipeline.value, table: table.value, contentType: contentType.value }
     },
   })
 

src/hooks/ingest.ts

@@ -87,6 +87,42 @@ const DASHBOARD: AppRouteRecordRaw = {
             },
           ],
         },
+        {
+          path: 'log-ingestion',
+          name: 'log-ingestion',
+          redirect: '/dashboard/ingest/log-ingestion/input',
+          component: () => import('@/views/dashboard/ingest/log-ingestion/index.vue'),
+          meta: {
+            locale: 'menu.dashboard.log-ingestion',
+            requiresAuth: false,
+            icon: 'upload-logs',
+            roles: ['admin', 'cloud'],
+          },
+          children: [
+            {
+              path: 'input',
+              name: 'log-ingestion-input',
+              component: () => import('@/views/dashboard/ingest/log-ingestion/input-log-ingestion.vue'),
+              meta: {
+                locale: 'menu.dashboard.input',
+                requiresAuth: false,
+                roles: ['admin', 'cloud'],
+                icon: 'input',
+              },
+            },
+            {
+              path: 'upload',
+              name: 'log-ingestion-upload',
+              component: () => import('@/views/dashboard/ingest/log-ingestion/upload-log-ingestion.vue'),
+              meta: {
+                locale: 'menu.dashboard.upload',
+                requiresAuth: false,
+                roles: ['admin', 'cloud'],
+                icon: 'upload',
+              },
+            },
+          ],
+        },
       ],
     },
     {

src/router/routes/modules/dashboard.ts

@@ -3,6 +3,7 @@ import { defineStore } from 'pinia'
 import Message from '@arco-design/web-vue/es/message'
 import i18n from '@/locale'
 import editorAPI from '@/api/editor'
+import { postPipelineLogs } from '@/api/pipeline'
 import dayjs from 'dayjs'
 import { dateTypes } from '@/views/dashboard/config'
 import { AnyObject } from '@/types/global'
@@ -205,14 +206,24 @@ const useCodeRunStore = defineStore('codeRun', () => {
     }
   }
 
+  const processLogs = async (data: string, table: string, pipeline: string, contentType: string) => {
+    try {
+      const appStore = useAppStore()
+      const res = await postPipelineLogs(appStore.database, table, pipeline, data, contentType)
+
+      return res
+    } catch (error: any) {
+      return error
+    }
+  }
+
   const runWithFormat = async (code: string, queryType: string, promForm?: PromForm, format?: string) => {
     const res: any =
       queryType === 'sql'
         ? await editorAPI.runSQLWithCSV(code, format)
         : await editorAPI.runPromQL(code, promForm, format)
     return res
   }
-
   return {
     results,
     runCode,
@@ -223,6 +234,7 @@ const useCodeRunStore = defineStore('codeRun', () => {
     explainResultKeyCount,
     explainResult,
     runWithFormat,
+    processLogs,
   }
 })
 export default useCodeRunStore

src/store/modules/code-run/index.ts

@@ -1,20 +1,52 @@
-import { ref } from 'vue'
-import { defineStore } from 'pinia'
+import { list as listPipelines } from '@/api/pipeline'
 
 const useIngestStore = defineStore('ingest', () => {
   const activeTab = ref('influxdb-input')
   const precision = ref('ns')
+  const pipelineName = ref('')
+  const tableForPipeline = ref('')
+  const contentType = ref('application/x-ndjson')
+  const pipelineList = ref([])
+  const pipelineLoading = ref(false)
+
   const footer = ref<{ [key: string]: boolean }>({
     'influxdb-input': true,
     'influxdb-upload': true,
     'log-ingestion-input': true,
     'log-ingestion-upload': true,
   })
 
+  const pipelineOptions = computed(() => {
+    return pipelineList.value.map((pipeline) => ({
+      label: pipeline.name,
+      value: pipeline.name,
+    }))
+  })
+
+  const fetchPipelines = async () => {
+    pipelineLoading.value = true
+    try {
+      pipelineList.value = await listPipelines()
+    } catch (error) {
+      console.error('Failed to fetch pipelines:', error)
+      pipelineList.value = []
+    } finally {
+      pipelineLoading.value = false
+    }
+  }
+
   return {
     activeTab,
     precision,
     footer,
+    pipelineName,
+    tableForPipeline,
+    contentType,
+    pipelineList,
+    pipelineLoading,
+    pipelineOptions,
+    fetchPipelines,
   }
 })
+
 export default useIngestStore

src/store/modules/ingest/index.ts

@@ -42,6 +42,7 @@ a-drawer.ingest(
 <script lang="ts" setup>
   import { Codemirror as CodeMirror } from 'vue-codemirror'
   import { basicSetup } from 'codemirror'
+  import { json } from '@codemirror/lang-json'
   import type { Log } from '@/store/modules/log/types'
 
   const props = defineProps({
@@ -59,7 +60,16 @@ a-drawer.ingest(
   const content = ref('')
   const docVisible = ref(false)
   const style = { height: '100%' }
-  const extensions = [basicSetup]
+
+  const extensions = computed(() => {
+    const contentType = props.config?.params?.contentType
+
+    if (contentType === 'application/json' || contentType === 'application/x-ndjson') {
+      return [basicSetup, json()]
+    }
+
+    return [basicSetup]
+  })
 
   const toggleDoc = () => {
     docVisible.value = !docVisible.value

src/views/dashboard/ingest/components/base-input.vue

@@ -66,7 +66,15 @@ a-layout-content.main-content
 
     a-spin(style="width: 100%" :tip="config.readingTip || 'Reading file...'" :loading="isReadingFile")
       a-card.file-scrollbar(:bordered="false")
-        CodeMirror(v-if="dataFromFile" v-model="codeInEditor" :disabled="true")
+        CodeMirror(
+          v-model="codeInEditor"
+          :style="{ height: '100%' }"
+          :extensions="extensions"
+          :spellcheck="true"
+          :indent-with-tab="true"
+          :tabSize="2"
+          :disabled="true"
+        )
       span.load(v-if="collapsed && remainingLines")
         a.text ...{{ remainingLines }} lines more
         a.button(type="text" size="mini" @click="loadMore") {{ config.expandText || 'Expand' }}
@@ -85,6 +93,7 @@ a-layout-content.main-content
 <script lang="ts" setup>
   import { Codemirror as CodeMirror } from 'vue-codemirror'
   import { basicSetup } from 'codemirror'
+  import { json } from '@codemirror/lang-json' // 导入JSON语法支持
   import type { Log } from '@/store/modules/log/types'
   import { isObject } from '@/utils/is'
 
@@ -218,6 +227,16 @@ a-layout-content.main-content
     footer.value[activeTab.value] = false
     isProcessLoading.value = false
   }
+
+  const extensions = computed(() => {
+    const contentType = props.config?.params?.contentType
+
+    if (contentType === 'application/json' || contentType === 'application/x-ndjson') {
+      return [basicSetup, json()]
+    }
+
+    return [basicSetup]
+  })
 </script>
 
 <style lang="less" scoped>

src/views/dashboard/ingest/components/base-upload.vue

@@ -47,6 +47,8 @@ a-space.top-bar
     padding-right: 20px;
     height: 58px;
     background: var(--card-bg-color);
+    width: 100%;
+
     :deep(.arco-select-view-value) {
       font-size: 13px;
       color: var(--main-font-color);