chore(metric): adopt pipeline release data format and support uid loo…

…kup (#114) Because - support pipeline release data format for metric data - allow resource id querying in metric endpoints This commit - support pipeline release data format for metric data - support resource uid lookup Resolves INS-1734 Resolves INS-1729
instill-ai · Sep 6, 2023 · 2bcbd2d · 2bcbd2d
1 parent 9707688
commit 2bcbd2d
Show file tree

Hide file tree

Showing 14 changed files with 441 additions and 156 deletions.
diff --git a/cmd/main/main.go b/cmd/main/main.go
@@ -137,12 +137,22 @@ func main() {
 		grpcServerOpts = append(grpcServerOpts, grpc.Creds(creds))
 	}
 
+	connectorPublicServiceClient, connectorPublicServiceClientConn := external.InitConnectorPublicServiceClient(ctx)
+	if connectorPublicServiceClientConn != nil {
+		defer connectorPublicServiceClientConn.Close()
+	}
+
+	pipelinePublicServiceClient, pipelinePublicServiceClientConn := external.InitPipelinePublicServiceClient(ctx)
+	if pipelinePublicServiceClientConn != nil {
+		defer pipelinePublicServiceClientConn.Close()
+	}
+
 	influxDBClient, influxDBQueryAPI := external.InitInfluxDBServiceClientV2(ctx, &config.Config)
 	defer influxDBClient.Close()
 
 	influxDB := repository.NewInfluxDB(influxDBQueryAPI, config.Config.InfluxDB.Bucket)
 	repository := repository.NewRepository(db)
-	service := service.NewService(repository, influxDB)
+	service := service.NewService(repository, influxDB, connectorPublicServiceClient, pipelinePublicServiceClient)
 
 	// Start usage reporter
 	var usg usage.Usage

diff --git a/config/config.go b/config/config.go
@@ -18,10 +18,12 @@ var Config AppConfig
 
 // AppConfig defines
 type AppConfig struct {
-	Server   ServerConfig   `koanf:"server"`
-	Database DatabaseConfig `koanf:"database"`
-	Log      LogConfig      `koanf:"log"`
-	InfluxDB InfluxDBConfig `koanf:"influxdb"`
+	Server           ServerConfig           `koanf:"server"`
+	Database         DatabaseConfig         `koanf:"database"`
+	Log              LogConfig              `koanf:"log"`
+	InfluxDB         InfluxDBConfig         `koanf:"influxdb"`
+	ConnectorBackend ConnectorBackendConfig `koanf:"connectorbackend"`
+	PipelineBackend  PipelineBackendConfig  `koanf:"pipelinebackend"`
 }
 
 // ServerConfig defines HTTP server configurations
@@ -43,6 +45,28 @@ type ServerConfig struct {
 	DefualtUserUid string `koanf:"defaultuseruid"`
 }
 
+// ConnectorBackendConfig related to connector-backend
+type ConnectorBackendConfig struct {
+	Host        string `koanf:"host"`
+	PrivatePort int    `koanf:"privateport"`
+	PublicPort  int    `koanf:"publicport"`
+	HTTPS       struct {
+		Cert string `koanf:"cert"`
+		Key  string `koanf:"key"`
+	}
+}
+
+// PipelineBackendConfig related to pipeline-backend
+type PipelineBackendConfig struct {
+	Host        string `koanf:"host"`
+	PrivatePort int    `koanf:"privateport"`
+	PublicPort  int    `koanf:"publicport"`
+	HTTPS       struct {
+		Cert string `koanf:"cert"`
+		Key  string `koanf:"key"`
+	}
+}
+
 // DatabaseConfig related to database
 type DatabaseConfig struct {
 	Username string `koanf:"username"`

diff --git a/config/config.yaml b/config/config.yaml
@@ -12,6 +12,20 @@ server:
     port: 443
   debug: true
   defaultuseruid:
+connectorbackend:
+  host: connector-backend
+  privateport: 3082
+  publicport: 8082
+  https:
+    cert:
+    key:
+pipelinebackend:
+  host: pipeline-backend
+  privateport: 3081
+  publicport: 8081
+  https:
+    cert:
+    key:
 database:
   username: postgres
   password: password

diff --git a/go.mod b/go.mod
@@ -12,7 +12,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0
 	github.com/iancoleman/strcase v0.2.0
 	github.com/influxdata/influxdb-client-go/v2 v2.12.3
-	github.com/instill-ai/protogen-go v0.3.3-alpha.0.20230814104042-37ca0356defc
+	github.com/instill-ai/protogen-go v0.3.3-alpha.0.20230905171903-a677797e7d2b
 	github.com/instill-ai/usage-client v0.2.4-alpha.0.20230814155646-874e57a1e4b0
 	github.com/instill-ai/x v0.3.0-alpha
 	github.com/jackc/pgx/v5 v5.2.0
@@ -39,6 +39,7 @@ require (
 )
 
 require (
+	cloud.google.com/go/longrunning v0.4.1 // indirect
 	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/apache/arrow/go/v12 v12.0.0 // indirect
 	github.com/apache/thrift v0.16.0 // indirect

diff --git a/go.sum b/go.sum
@@ -38,6 +38,8 @@ cloud.google.com/go/bigquery v1.8.0/go.mod h1:J5hqkt3O0uAFnINi6JXValWIb1v0goeZM7
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
 cloud.google.com/go/firestore v1.1.0/go.mod h1:ulACoGHTpvq5r8rxGJ4ddJZBZqakUQqClKRT5SZwBmk=
+cloud.google.com/go/longrunning v0.4.1 h1:v+yFJOfKC3yZdY6ZUI933pIYdhyhV8S3NpWrXWmg7jM=
+cloud.google.com/go/longrunning v0.4.1/go.mod h1:4iWDqhBZ70CvZ6BfETbvam3T8FMvLK+eFj0E6AaRQTo=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
 cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
@@ -736,8 +738,8 @@ github.com/influxdata/line-protocol/v2 v2.0.0-20210312151457-c52fdecb625a/go.mod
 github.com/influxdata/line-protocol/v2 v2.1.0/go.mod h1:QKw43hdUBg3GTk2iC3iyCxksNj7PX9aUSeYOYE/ceHY=
 github.com/influxdata/line-protocol/v2 v2.2.1 h1:EAPkqJ9Km4uAxtMRgUubJyqAr6zgWM0dznKMLRauQRE=
 github.com/influxdata/line-protocol/v2 v2.2.1/go.mod h1:DmB3Cnh+3oxmG6LOBIxce4oaL4CPj3OmMPgvauXh+tM=
-github.com/instill-ai/protogen-go v0.3.3-alpha.0.20230814104042-37ca0356defc h1:1akaOOUCsVU2yzFIb5KwVw5GRy7v+hyNmNmJD+2x+l8=
-github.com/instill-ai/protogen-go v0.3.3-alpha.0.20230814104042-37ca0356defc/go.mod h1:qsq5ecnA1xi2rLnVQFo/9xksA7I7wQu8c7rqM5xbIrQ=
+github.com/instill-ai/protogen-go v0.3.3-alpha.0.20230905171903-a677797e7d2b h1:UCJ7uXxgqLXGnRcJeMp73YI6SHszkRS8dog/j9vHI6E=
+github.com/instill-ai/protogen-go v0.3.3-alpha.0.20230905171903-a677797e7d2b/go.mod h1:z/L84htamlJ4QOR4jtJOaa+y3Hihu7WEqOipW0LEkmc=
 github.com/instill-ai/usage-client v0.2.4-alpha.0.20230814155646-874e57a1e4b0 h1:9QoCxaktvqGJYGjN8KhkWsv1DVfwbt5G1d/Ycx1kJxo=
 github.com/instill-ai/usage-client v0.2.4-alpha.0.20230814155646-874e57a1e4b0/go.mod h1:SELFgirs+28Wfnh0kGw02zttit4pUeKLKp17zGsTu6g=
 github.com/instill-ai/x v0.3.0-alpha h1:z9fedROOG2dVHhswBfVwU/hzHuq8/JKSUON7inF+FH8=

diff --git a/integration-test/grpc-public-user.js b/integration-test/grpc-public-user.js
@@ -245,6 +245,9 @@ export function CheckPublicDeleteToken() {
 
 export function CheckPublicMetrics() {
 
+  let pipeline_id = randomString(10)
+  let connector_id = randomString(10)
+
   client.connect(constant.mgmtPublicGRPCHost, {
     plaintext: true
   });
@@ -264,7 +267,7 @@ export function CheckPublicMetrics() {
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerRecords response has next_page_token': (r) => r && r.message.nextPageToken !== undefined,
     });
     check(client.invoke('base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerRecords', {
-      filter: "pipeline_id=\"a\" AND trigger_mode=MODE_SYNC",
+      filter: `pipeline_id="${pipeline_id}" AND trigger_mode=MODE_SYNC`,
     }), {
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerRecords with filter status': (r) => r && r.status == grpc.StatusOK,
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerRecords with filter response pipelineTriggerRecords length is 0': (r) => r && r.message.pipelineTriggerRecords.length === 0,
@@ -289,7 +292,7 @@ export function CheckPublicMetrics() {
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerTableRecords response has next_page_token': (r) => r && r.message.nextPageToken !== undefined,
     });
     check(client.invoke('base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerTableRecords', {
-      filter: "pipeline_id=\"iloveinstill\"",
+      filter: `pipeline_id="${pipeline_id}"`,
     }), {
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerTableRecords with filter status': (r) => r && r.status == grpc.StatusOK,
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerTableRecords with filter response pipelineTriggerTableRecords length is 0': (r) => r && r.message.pipelineTriggerTableRecords.length === 0,
@@ -306,7 +309,7 @@ export function CheckPublicMetrics() {
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerChartRecords response has pipelineTriggerChartRecords': (r) => r && r.message.pipelineTriggerChartRecords !== undefined,
     });
     check(client.invoke('base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerChartRecords', {
-      filter: "pipeline_id=\"a\" AND trigger_mode=MODE_SYNC",
+      filter: `pipeline_id="${pipeline_id}" AND trigger_mode=MODE_SYNC`,
     }), {
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerChartRecords with filter status': (r) => r && r.status == grpc.StatusOK,
       'base.mgmt.v1alpha.MgmtPublicService/ListPipelineTriggerChartRecords with filter response pipelineTriggerChartRecords lenght is 0': (r) => r && r.message.pipelineTriggerChartRecords.length === 0,
@@ -329,7 +332,7 @@ export function CheckPublicMetrics() {
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteRecords response has next_page_token': (r) => r && r.message.nextPageToken !== undefined,
     });
     check(client.invoke('base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteRecords', {
-      filter: "connector_id=\"a\" AND status=STATUS_COMPLETED",
+      filter: `connector_id="${connector_id}" AND status=STATUS_COMPLETED`,
     }), {
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteRecords with filter status': (r) => r && r.status == grpc.StatusOK,
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteRecords with filter response connectorExecuteRecords length is 0': (r) => r && r.message.connectorExecuteRecords.length === 0,
@@ -354,7 +357,7 @@ export function CheckPublicMetrics() {
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteTableRecords response has next_page_token': (r) => r && r.message.nextPageToken !== undefined,
     });
     check(client.invoke('base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteTableRecords', {
-      filter: "connector_id=\"iloveinstill\"",
+      filter: `connector_id="${connector_id}"`,
     }), {
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteTableRecords with filter status': (r) => r && r.status == grpc.StatusOK,
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteTableRecords with filter response connectorExecuteTableRecords length is 0': (r) => r && r.message.connectorExecuteTableRecords.length === 0,
@@ -371,7 +374,7 @@ export function CheckPublicMetrics() {
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteChartRecords response has connectorExecuteChartRecords': (r) => r && r.message.connectorExecuteChartRecords !== undefined,
     });
     check(client.invoke('base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteChartRecords', {
-      filter: "connector_id=\"a\" AND status=STATUS_COMPLETED",
+      filter: `connector_id="${connector_id}" AND status=STATUS_COMPLETED`,
     }), {
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteChartRecords with filter status': (r) => r && r.status == grpc.StatusOK,
       'base.mgmt.v1alpha.MgmtPublicService/ListConnectorExecuteChartRecords with filter response connectorExecuteChartRecords lenght is 0': (r) => r && r.message.connectorExecuteChartRecords.length === 0,

diff --git a/integration-test/rest-public-user.js b/integration-test/rest-public-user.js
@@ -274,6 +274,8 @@ export function CheckPublicDeleteToken() {
 export function CheckPublicMetrics() {
   group(`Management Public API: List Pipeline Trigger Records`, () => {
 
+    let pipeline_id = randomString(10)
+
     let emptyPipelineTriggerRecordResponse = {
       "pipelineTriggerRecords": [],
       "nextPageToken": "",
@@ -299,7 +301,7 @@ export function CheckPublicMetrics() {
     check(
       http.request(
         "GET",
-        `${constant.mgmtPublicHost}/metrics/vdp/pipeline/triggers?filter=trigger_mode=MODE_SYNC%20AND%20pipeline_id=%22a%22`
+        `${constant.mgmtPublicHost}/metrics/vdp/pipeline/triggers?filter=trigger_mode=MODE_SYNC%20AND%20pipeline_id=%22${pipeline_id}%22`
       ),
       {
         [`GET /${constant.mgmtVersion}/metrics/vdp/pipeline/triggers with filter response status is 200`]:
@@ -315,6 +317,8 @@ export function CheckPublicMetrics() {
   })
   group(`Management Public API: List Pipeline Trigger Table Records`, () => {
 
+    let pipeline_id = randomString(10)
+
     let emptyPipelineTriggerTableRecordResponse = {
       "pipelineTriggerTableRecords": [],
       "nextPageToken": "",
@@ -340,7 +344,7 @@ export function CheckPublicMetrics() {
     check(
       http.request(
         "GET",
-        `${constant.mgmtPublicHost}/metrics/vdp/pipeline/tables?filter=pipeline_id=%22iloveinstill%22`
+        `${constant.mgmtPublicHost}/metrics/vdp/pipeline/tables?filter=pipeline_id=%22${pipeline_id}%22`
       ),
       {
         [`GET /${constant.mgmtVersion}/metrics/vdp/pipeline/tables with filter response status is 200`]:
@@ -355,6 +359,9 @@ export function CheckPublicMetrics() {
     )
   })
   group(`Management Public API: List Pipeline Trigger Chart Records`, () => {
+
+    let pipeline_id = randomString(10)
+
     check(
       http.request(
         "GET",
@@ -370,7 +377,7 @@ export function CheckPublicMetrics() {
     check(
       http.request(
         "GET",
-        `${constant.mgmtPublicHost}/metrics/vdp/pipeline/charts?filter=trigger_mode=MODE_SYNC%20AND%20pipeline_id=%22a%22`
+        `${constant.mgmtPublicHost}/metrics/vdp/pipeline/charts?filter=trigger_mode=MODE_SYNC%20AND%20pipeline_id=%22${pipeline_id}%22`
       ),
       {
         [`GET /${constant.mgmtVersion}/metrics/vdp/pipeline/charts with filter response status is 200`]:
@@ -382,6 +389,8 @@ export function CheckPublicMetrics() {
   })
   group(`Management Public API: List Connector Execute Records`, () => {
 
+    let connector_id = randomString(10)
+
     let emptyConnectorExecuteRecordResponse = {
       "connectorExecuteRecords": [],
       "nextPageToken": "",
@@ -407,7 +416,7 @@ export function CheckPublicMetrics() {
     check(
       http.request(
         "GET",
-        `${constant.mgmtPublicHost}/metrics/vdp/connector/executes?filter=status=STATUS_COMPLETED%20AND%20connector_id=%22a%22`
+        `${constant.mgmtPublicHost}/metrics/vdp/connector/executes?filter=status=STATUS_COMPLETED%20AND%20connector_id=%22${connector_id}%22`
       ),
       {
         [`GET /${constant.mgmtVersion}/metrics/vdp/connector/executes with filter response status is 200`]:
@@ -423,6 +432,8 @@ export function CheckPublicMetrics() {
   })
   group(`Management Public API: List Connector Execute Table Records`, () => {
 
+    let connector_id = randomString(10)
+
     let emptyConnectorExecuteTableRecordResponse = {
       "connectorExecuteTableRecords": [],
       "nextPageToken": "",
@@ -448,7 +459,7 @@ export function CheckPublicMetrics() {
     check(
       http.request(
         "GET",
-        `${constant.mgmtPublicHost}/metrics/vdp/connector/tables?filter=connector_id=%22iloveinstill%22`
+        `${constant.mgmtPublicHost}/metrics/vdp/connector/tables?filter=connector_id=%22${connector_id}%22`
       ),
       {
         [`GET /${constant.mgmtVersion}/metrics/vdp/connector/tables with filter response status is 200`]:
@@ -463,6 +474,9 @@ export function CheckPublicMetrics() {
     )
   })
   group(`Management Public API: List Connector Execute Chart Records`, () => {
+
+    let connector_id = randomString(10)
+
     check(
       http.request(
         "GET",
@@ -478,7 +492,7 @@ export function CheckPublicMetrics() {
     check(
       http.request(
         "GET",
-        `${constant.mgmtPublicHost}/metrics/vdp/connector/charts?filter=status=STATUS_COMPLETED%20AND%20connector_id=%22a%22`
+        `${constant.mgmtPublicHost}/metrics/vdp/connector/charts?filter=status=STATUS_COMPLETED%20AND%20connector_id=%22${connector_id}%22`
       ),
       {
         [`GET /${constant.mgmtVersion}/metrics/vdp/connector/charts with filter response status is 200`]:

diff --git a/pkg/constant/constant.go b/pkg/constant/constant.go
@@ -18,3 +18,33 @@ const HeaderUserUIDKey = "jwt-sub"
 
 // MaxApiKeyNum is the maximum number of API keys
 const MaxApiKeyNum = 10
+
+// Filter enum
+const (
+	Start              string = "start"
+	Stop               string = "stop"
+	PipelineID         string = "pipeline_id"
+	PipelineUID        string = "pipeline_uid"
+	PipelineReleaseID  string = "pipeline_release_id"
+	PipelineReleaseUID string = "pipeline_release_uid"
+	ConnectorID        string = "connector_id"
+	ConnectorUID       string = "connector_uid"
+	TriggerMode        string = "trigger_mode"
+	Status             string = "status"
+)
+
+// Metric data enum
+const (
+	OwnerUID                    string = "owner_uid"
+	PipelineTriggerMeasurement  string = "pipeline.trigger"
+	ConnectorExecuteMeasurement string = "connector.execute"
+	PipelineMode                string = "pipeline_mode"
+	PipelineTriggerID           string = "pipeline_trigger_id"
+	ConnectorExecuteID          string = "connector_execute_id"
+	ConnectorDefinitionUID      string = "connector_definition_uid"
+	TriggerTime                 string = "trigger_time"
+	Executetime                 string = "execute_time"
+	ComputeTimeDuration         string = "compute_time_duration"
+	Completed                   string = "completed"
+	Errored                     string = "errored"
+)