list metric run

lhctl/internal/metrics.go

@@ -44,6 +44,33 @@ func toType(metricType string) lhproto.MetricType {
 		panic("Unrecognized metric type " + metricType)
 	}
 }
+
+var listMetricRuns = &cobra.Command{
+	Use:   "metricRun <wfRunId>",
+	Short: "List all MetricRun's for a given Metric Id.",
+	Long:  ``,
+	Args:  cobra.ExactArgs(2),
+	Run: func(cmd *cobra.Command, args []string) {
+		measurable := args[0]
+		metricType := args[1]
+
+		metricId := &lhproto.MetricId{
+			Measurable: toMeasurable(measurable),
+			Type:       toType(metricType),
+		}
+
+		req := &lhproto.ListMetricRunRequest{
+			MetricId: metricId,
+		}
+
+		littlehorse.PrintResp(getGlobalClient(cmd).ListMetricRuns(
+			requestContext(cmd),
+			req,
+		))
+	},
+}
+
 func init() {
 	putCmd.AddCommand(putMetricCmd)
+	listCmd.AddCommand(listMetricRuns)
 }

local-dev/configs/server-1.config

@@ -2,7 +2,7 @@ LHS_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
 LHS_CLUSTER_ID=my-cluster
 LHS_INSTANCE_ID=1
 LHS_REPLICATION_FACTOR=1
-LHS_CLUSTER_PARTITIONS=1
+LHS_CLUSTER_PARTITIONS=21
 LHS_STATE_DIR=/tmp/kafkaState
 LHS_STREAMS_NUM_WARMUP_REPLICAS=8
 LHS_STREAMS_NUM_STANDBY_REPLICAS=0

schemas/internal/command.proto

@@ -113,8 +113,9 @@ message AggregateTaskMetrics {
 }
 
 message AggregateMetrics {
-    repeated RepartitionWindowedMetric windowed_metrics = 1;
-    TenantId tenant_id = 2;
+    MetricId metric_id = 1;
+    repeated RepartitionWindowedMetric windowed_metrics = 2;
+    TenantId tenant_id = 3;
 }
 
 message WfMetricUpdate {
@@ -259,7 +260,6 @@ message DeleteTaskWorkerGroupRequest {
 }
 
 message RepartitionWindowedMetric {
-    MetricId metric_id = 1;
-    double value = 2;
-    google.protobuf.Timestamp window_start = 3;
+    double value = 1;
+    google.protobuf.Timestamp window_start = 2;
 }

schemas/littlehorse/service.proto

@@ -296,6 +296,8 @@ service LittleHorse {
 
   // Gets the version of the LH Server.
   rpc GetServerVersion(google.protobuf.Empty) returns (ServerVersion) {}
+
+  rpc ListMetricRuns(ListMetricRunRequest) returns (MetricRunList) {}
 }
 
 // Returns the UserTaskDef with a given name and the highest version number.
@@ -1450,3 +1452,12 @@ message ServerVersion {
   // but rather a release candidate or experimental pre-release.
   optional string pre_release_identifier = 4;
 }
+
+
+message MetricRunList {
+  repeated MetricRun results = 1;
+}
+
+message ListMetricRunRequest {
+  MetricId metric_id = 1;
+}