feat: log query and log pipelines

.eslintrc.js

@@ -69,5 +69,6 @@ module.exports = {
     'import/no-extraneous-dependencies': 0,
     'noUnusedLocals': 0,
     'prefer-destructuring': ['error', { object: true, array: false }],
+    'no-continue': 1,
   },
 }

package.json

@@ -32,7 +32,7 @@
     ]
   },
   "dependencies": {
-    "@arco-design/web-vue": "^2.41.0",
+    "@arco-design/web-vue": "^2.56.2",
     "@babel/core": "^7.20.5",
     "@codemirror/autocomplete": "^6.4.2",
     "@codemirror/lang-java": "^6.0.1",
@@ -84,7 +84,8 @@
     "vue-codemirror": "^6.1.1",
     "vue-echarts": "^6.5.0",
     "vue-i18n": "^9.2.2",
-    "vue-router": "^4.0.14"
+    "vue-router": "^4.0.14",
+    "js-file-download": "^0.4.12"
   },
   "devDependencies": {
     "@babel/types": "^7.21.4",

src/api/editor.ts

@@ -2,6 +2,7 @@ import axios, { AxiosRequestConfig, AxiosRequestHeaders } from 'axios'
 import dayjs from 'dayjs'
 import qs from 'qs'
 import { PromForm } from '@/store/modules/code-run/types'
+import { HttpResponse } from './interceptor'
 
 const sqlUrl = `/v1/sql`
 const scriptUrl = `/v1/scripts`
@@ -97,7 +98,7 @@ const getTableByName = (tableName: string) => {
   )
 }
 
-const runSQL = (code: string) => {
+const runSQL = (code: string): Promise<HttpResponse> => {
   return axios.post(sqlUrl, makeSqlData(code), addDatabaseParams())
 }
 
@@ -136,6 +137,11 @@ const writeInfluxDB = (data: string, precision: string) => {
   } as AxiosRequestConfig
   return axios.post(`${influxURL}/write`, data, config)
 }
+const runSQLWithCSV = (code: string): Promise<HttpResponse> => {
+  const params = addDatabaseParams()
+  params.params.format = 'csv'
+  return axios.post(sqlUrl, makeSqlData(code), params)
+}
 
 export default {
   getTables,
@@ -149,4 +155,5 @@ export default {
   writeInfluxDB,
   checkScriptsTable,
   fetchTablesCount,
+  runSQLWithCSV,
 }

src/api/interceptor.ts

@@ -23,6 +23,15 @@ export interface Auth {
 }
 
 // todo: can we use env and proxy at the same time?
+export const TableNameReg = /(?<=from|FROM)\s+([^\s;]+)/
+export function parseTable(sql: string) {
+  const result = sql.match(TableNameReg)
+  if (result && result.length) {
+    const arr = result[1].trim().split('.')
+    return arr[arr.length - 1]
+  }
+  return ''
+}
 
 axios.interceptors.request.use(
   (config: AxiosRequestConfig) => {
@@ -51,6 +60,7 @@ axios.interceptors.request.use(
     return Promise.reject(error)
   }
 )
+const ignoreList = ['pipelines']
 
 axios.interceptors.response.use(
   (response: AxiosResponse) => {
@@ -70,16 +80,22 @@ axios.interceptors.response.use(
       return Promise.reject(errorResponse)
     }
     if (isV1) {
+      if (response.config.params && response.config.params.format === 'csv') {
+        return response.data
+      }
       response.data = JSONbigint({ storeAsString: true }).parse(response.data)
       const { data } = response
       if (data.code && data.code !== 0) {
         // v1 and error
-        Message.error({
-          content: data.error || 'Error',
-          duration: 5 * 1000,
-          closable: true,
-          resetOnHover: true,
-        })
+        const tableName = parseTable(decodeURIComponent(response.config.data))
+        if (ignoreList.indexOf(tableName) === -1) {
+          Message.error({
+            content: data.error || 'Error',
+            duration: 5 * 1000,
+            closable: true,
+            resetOnHover: true,
+          })
+        }
         const error = {
           error: data.error || 'Error',
           startTime: new Date(response.config.traceTimeStart).toLocaleTimeString(),
@@ -104,7 +120,8 @@ axios.interceptors.response.use(
     }
     const data = JSON.parse(error.response.data)
     const isInflux = !!error.config.url?.startsWith(`/v1/influxdb`)
-    if (!isInflux) {
+    const tableName = parseTable(decodeURIComponent(error.response.config.data))
+    if (!isInflux && ignoreList.indexOf(tableName) === -1) {
       Message.error({
         content: data.error || 'Request Error',
         duration: 5 * 1000,

src/api/pipeline.ts

@@ -0,0 +1,92 @@
+import axios from 'axios'
+import qs from 'qs'
+import dayjs from 'dayjs'
+import editorAPI from './editor'
+
+const { runSQL } = editorAPI
+const url = '/v1/events/pipelines'
+const sqlUrl = `/v1/sql`
+
+const makeSqlData = (sql: string) => {
+  return qs.stringify({
+    sql,
+  })
+}
+
+export type PipeFile = {
+  name: string
+  content?: string
+  version: string
+}
+
+/* eslint-disable */
+function nanoTimestampToUTCString(nanoTimestamp: bigint) {
+  nanoTimestamp = BigInt(nanoTimestamp)
+  const divide1 = BigInt(1000000000)
+  const seconds = nanoTimestamp / divide1 // 纳秒转秒
+  const divide2 = BigInt(1000000)
+  const milliseconds = Number((nanoTimestamp % divide1) / divide2) // 毫秒部分
+  const nanoseconds = Number(nanoTimestamp % divide2) // 纳秒部分
+
+  // 使用 Date 对象将秒转换为 UTC
+  const date = new Date(Number(seconds) * 1000)
+  const isoString = date.toISOString().replace('Z', '').replace('.000', '')
+  // 拼接毫秒和纳秒部分
+  return `${isoString}.${String(milliseconds).padStart(3, '0')}${String(nanoseconds).padStart(6, '0')}Z`
+}
+
+/* eslint-enable */
+export function create(pipeFile: PipeFile) {
+  const { content, name } = pipeFile
+  const file = new File([content], `${name}.yaml`, {
+    type: 'application/yaml',
+  })
+  const formData = new FormData()
+  formData.append('file', file)
+  return axios.postForm(`${url}/${name}`, formData)
+}
+
+export function list() {
+  return runSQL(
+    `SELECT name,max(created_at) as created_at FROM greptime_private.pipelines group by name order by created_at desc`
+  ).then((result) => {
+    return result.output[0].records.rows.map((row: any) => {
+      return {
+        name: row[0],
+        version: nanoTimestampToUTCString(row[1]),
+      }
+    })
+  })
+}
+
+export function getByName(name: string): Promise<PipeFile> {
+  const sql = `select name, created_at, pipeline 
+    from greptime_private.pipelines 
+    where name = '${name}'
+    order by created_at desc 
+    limit 1`
+  return runSQL(sql).then((result) => {
+    const row = result.output[0].records.rows[0] as any
+    return {
+      name: row[0],
+      version: nanoTimestampToUTCString(row[1]),
+      content: row[2],
+    }
+  })
+}
+
+export function del(name: string, version: string) {
+  return axios.delete(`${url}/${name}`, {
+    params: {
+      version,
+    },
+  })
+}
+
+export function debug(name: string, content: any) {
+  return axios.post(`${url}/dryrun?pipeline_name=${name}`, JSON.parse(content), {
+    headers: {
+      'Content-Type': 'application/json', // Set Content-Type
+    },
+  })
+}

src/assets/style/select.less

@@ -2,7 +2,7 @@
   padding-right: 8px;
   padding-left: 8px;
   background: var(--grey-bg-color);
-  border-radius: 4px;
+  // border-radius: 4px;
 }
 
 .arco-select-view-single:hover {

src/components/index.ts

@@ -13,6 +13,7 @@ import Chart from './chart/index.vue'
 import Breadcrumb from './breadcrumb/index.vue'
 import 'echarts/lib/component/dataset'
 import 'echarts/lib/component/transform'
+import YMLEditor from './yml-editor.vue'
 
 // Manually introduce ECharts modules to reduce packing size
 
@@ -32,5 +33,6 @@ export default {
   install(Vue: App) {
     Vue.component('Chart', Chart)
     Vue.component('Breadcrumb', Breadcrumb)
+    Vue.component('YMLEditorSimple', YMLEditor)
   },
 }