[V2 API] Merger support for observation API (#1094)

Author: shifucun

cmd/main.go

@@ -68,6 +68,8 @@ var (
 	tmcfCsvBucket  = flag.String("tmcf_csv_bucket", "", "The GCS bucket that contains tmcf and csv files")
 	tmcfCsvFolder  = flag.String("tmcf_csv_folder", "", "GCS folder for an import. An import must have a unique prefix within a bucket.")
 	memdbPath      = flag.String("memdb_path", "", "File path of memdb config")
+	// Remote mixer url. The API serves merged data from local and remote mixer
+	remoteMixerURL = flag.String("remote_mixer_url", "", "Remote mixer url to fetch and merge data for API response.")
 	// Profile startup memory instead of listening for requests
 	startupMemoryProfile = flag.String("startup_memprof", "", "File path to write the memory profile of mixer startup to")
 	// Serve live profiles of the process (CPU, memory, etc.) over HTTP on this port
@@ -174,6 +176,7 @@ func main() {
 		*mixerProject,
 		*bqDataset,
 		*schemaPath,
+		*remoteMixerURL,
 	)
 	if err != nil {
 		log.Fatalf("Failed to create metadata: %v", err)

internal/merger/observation.go

@@ -0,0 +1,47 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package merger provides function to merge V2 API ressponses
+package merger
+
+import (
+	pbv2 "github.com/datacommonsorg/mixer/internal/proto/v2"
+)
+
+// MergeObservation merges two V2 observation response.
+func MergeObservation(
+	o1, o2 *pbv2.ObservationResponse) (*pbv2.ObservationResponse, error) {
+	for v, vData := range o2.ByVariable {
+		if _, ok := o1.ByVariable[v]; !ok {
+			o1.ByVariable[v] = &pbv2.VariableObservation{
+				ByEntity: map[string]*pbv2.EntityObservation{},
+			}
+		}
+		for e, eData := range vData.ByEntity {
+			if _, ok := o1.ByVariable[v].ByEntity[e]; !ok {
+				o1.ByVariable[v].ByEntity[e] = &pbv2.EntityObservation{
+					OrderedFacets: []*pbv2.FacetObservation{},
+				}
+			}
+			eData.OrderedFacets = append(
+				eData.OrderedFacets,
+				o1.ByVariable[v].ByEntity[e].OrderedFacets...,
+			)
+		}
+	}
+	for facetID, facet := range o2.Facets {
+		o1.Facets[facetID] = facet
+	}
+	return o1, nil
+}