add event-combine processor

.github/workflows/release.yml

@@ -46,7 +46,7 @@ jobs:
           password: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Release with goreleaser
-        uses: goreleaser/goreleaser-action@v4
+        uses: goreleaser/goreleaser-action@v5
         with:
           version: ${{ env.GORELEASER_VER }}
           args: release --rm-dist

app/api.go

@@ -288,7 +288,7 @@ func (a *App) handleClusteringGet(w http.ResponseWriter, r *http.Request) {
 }
 
 func (a *App) handleHealthzGet(w http.ResponseWriter, r *http.Request) {
-	s := map[string]string{"status": "healthy",}
+	s := map[string]string{"status": "healthy"}
 	b, err := json.Marshal(s)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)

app/clustering.go

@@ -670,7 +670,7 @@ func (a *App) unassignTarget(ctx context.Context, name string, serviceID string)
 		}
 		rsp, err := client.Do(req)
 		if err != nil {
-			rsp.Body.Close()
+			// don't close the body here since Body will be nil
 			a.Logger.Printf("failed HTTP request: %v", err)
 			continue
 		}

app/diff.go

@@ -19,13 +19,14 @@ import (
 	"strings"
 
 	"github.com/openconfig/gnmi/proto/gnmi"
-	"github.com/openconfig/gnmic/config"
-	"github.com/openconfig/gnmic/formatters"
-	"github.com/openconfig/gnmic/types"
 	"github.com/openconfig/grpctunnel/tunnel"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"google.golang.org/protobuf/proto"
+
+	"github.com/openconfig/gnmic/config"
+	"github.com/openconfig/gnmic/formatters"
+	"github.com/openconfig/gnmic/types"
 )
 
 type targetDiffResponse struct {

app/gnmi_client_subscribe.go

@@ -18,12 +18,13 @@ import (
 	"time"
 
 	"github.com/openconfig/gnmi/proto/gnmi"
+	"github.com/openconfig/grpctunnel/tunnel"
+	"google.golang.org/grpc"
+
 	"github.com/openconfig/gnmic/config"
 	"github.com/openconfig/gnmic/lockers"
 	"github.com/openconfig/gnmic/outputs"
 	"github.com/openconfig/gnmic/types"
-	"github.com/openconfig/grpctunnel/tunnel"
-	"google.golang.org/grpc"
 )
 
 type subscriptionRequest struct {
@@ -174,7 +175,7 @@ func (a *App) clientSubscribe(ctx context.Context, tc *types.TargetConfig) error
 	}
 	subRequests := make([]subscriptionRequest, 0, len(subscriptionsConfigs))
 	for scName, sc := range subscriptionsConfigs {
-		req, err := a.Config.CreateSubscribeRequest(sc, tc.Name)
+		req, err := a.Config.CreateSubscribeRequest(sc, tc)
 		if err != nil {
 			if errors.Is(errors.Unwrap(err), config.ErrConfig) {
 				fmt.Fprintf(os.Stderr, "%v\n", err)
@@ -243,7 +244,7 @@ func (a *App) clientSubscribeOnce(ctx context.Context, tc *types.TargetConfig) e
 	}
 	subRequests := make([]subscriptionRequest, 0)
 	for _, sc := range subscriptionsConfigs {
-		req, err := a.Config.CreateSubscribeRequest(sc, tc.Name)
+		req, err := a.Config.CreateSubscribeRequest(sc, tc)
 		if err != nil {
 			if errors.Is(errors.Unwrap(err), config.ErrConfig) {
 				fmt.Fprintf(os.Stderr, "%v\n", err)

app/metrics.go

@@ -69,7 +69,7 @@ func (a *App) startClusterMetrics() {
 			if err != nil {
 				a.Logger.Printf("failed to get leader key: %v", err)
 			}
-			if leader[leaderKey] == a.Config.InstanceName {
+			if leader[leaderKey] == a.Config.Clustering.InstanceName {
 				clusterIsLeader.Set(1)
 			} else {
 				clusterIsLeader.Set(0)
@@ -84,7 +84,7 @@ func (a *App) startClusterMetrics() {
 			}
 			numLockedNodes := 0
 			for _, v := range lockedNodes {
-				if v == a.Config.InstanceName {
+				if v == a.Config.Clustering.InstanceName {
 					numLockedNodes++
 				}
 			}

cache/oc_cache.go

@@ -23,8 +23,9 @@ import (
 	"github.com/openconfig/gnmi/path"
 	"github.com/openconfig/gnmi/proto/gnmi"
 	"github.com/openconfig/gnmi/subscribe"
-	"github.com/openconfig/gnmic/utils"
 	"google.golang.org/protobuf/proto"
+
+	"github.com/openconfig/gnmic/utils"
 )
 
 const (
@@ -298,13 +299,13 @@ func (gc *gnmiCache) handleOnChangeQuery(ctx context.Context, ro *ReadOpts, ch c
 	caches := gc.getCaches(ro.Subscription)
 	numCaches := len(caches)
 	gc.logger.Printf("on-change query got %d caches", numCaches)
+
 	wg := new(sync.WaitGroup)
 	wg.Add(numCaches)
 
 	for name, c := range caches {
 		go func(name string, c *subCache) {
 			defer wg.Done()
-
 			for _, p := range ro.Paths {
 				// handle updates only
 				if !ro.UpdatesOnly {
@@ -330,6 +331,9 @@ func (gc *gnmiCache) handleOnChangeQuery(ctx context.Context, ro *ReadOpts, ch c
 				}
 				// main on-change subscription
 				fp := path.ToStrings(p, true)
+				fp = append(fp, "")
+				copy(fp[1:], fp)
+				fp[0] = ro.Target
 				// set callback
 				mc := &matchClient{name: name, ch: ch}
 				remove := c.match.AddQuery(fp, mc)

cmd/prompt.go

@@ -21,10 +21,11 @@ import (
 	homedir "github.com/mitchellh/go-homedir"
 	"github.com/nsf/termbox-go"
 	"github.com/olekukonko/tablewriter"
-	"github.com/openconfig/gnmic/types"
 	"github.com/openconfig/goyang/pkg/yang"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+
+	"github.com/openconfig/gnmic/types"
 )
 
 var colorMapping = map[string]goprompt.Color{
@@ -298,13 +299,17 @@ func subscriptionTable(scs map[string]*types.SubscriptionConfig, list bool) [][]
 	if list {
 		tabData := make([][]string, 0, len(scs))
 		for _, sub := range scs {
+			enc := ""
+			if sub.Encoding != nil {
+				enc = *sub.Encoding
+			}
 			tabData = append(tabData, []string{
 				sub.Name,
 				sub.ModeString(),
 				sub.PrefixString(),
 				sub.PathsString(),
 				sub.SampleIntervalString(),
-				sub.Encoding,
+				enc,
 			})
 		}
 		sort.Slice(tabData, func(i, j int) bool {
@@ -323,7 +328,7 @@ func subscriptionTable(scs map[string]*types.SubscriptionConfig, list bool) [][]
 		tabData = append(tabData, []string{"Prefix", sub.PrefixString()})
 		tabData = append(tabData, []string{"Paths", sub.PathsString()})
 		tabData = append(tabData, []string{"Sample Interval", sub.SampleIntervalString()})
-		tabData = append(tabData, []string{"Encoding", sub.Encoding})
+		tabData = append(tabData, []string{"Encoding", *sub.Encoding})
 		tabData = append(tabData, []string{"Qos", sub.QosString()})
 		tabData = append(tabData, []string{"Heartbeat Interval", sub.HeartbeatIntervalString()})
 		return tabData
@@ -694,9 +699,11 @@ func subscriptionDescription(sub *types.SubscriptionConfig) string {
 			sb.WriteString(", ")
 		}
 	}
-	sb.WriteString("encoding=")
-	sb.WriteString(sub.Encoding)
-	sb.WriteString(", ")
+	if sub.Encoding != nil {
+		sb.WriteString("encoding=")
+		sb.WriteString(*sub.Encoding)
+		sb.WriteString(", ")
+	}
 	if sub.Prefix != "" {
 		sb.WriteString("prefix=")
 		sb.WriteString(sub.Prefix)

config/config.go

@@ -26,14 +26,15 @@ import (
 	"github.com/itchyny/gojq"
 	"github.com/mitchellh/go-homedir"
 	"github.com/openconfig/gnmi/proto/gnmi"
-	"github.com/openconfig/gnmic/api"
-	"github.com/openconfig/gnmic/types"
-	"github.com/openconfig/gnmic/utils"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 	"gopkg.in/natefinch/lumberjack.v2"
 	yaml "gopkg.in/yaml.v2"
+
+	"github.com/openconfig/gnmic/api"
+	"github.com/openconfig/gnmic/types"
+	"github.com/openconfig/gnmic/utils"
 )
 
 const (

config/diff.go

@@ -13,9 +13,10 @@ import (
 	"strings"
 
 	"github.com/openconfig/gnmi/proto/gnmi"
+	"github.com/spf13/cobra"
+
 	"github.com/openconfig/gnmic/api"
 	"github.com/openconfig/gnmic/types"
-	"github.com/spf13/cobra"
 )
 
 func (c *Config) CreateDiffSubscribeRequest(cmd *cobra.Command) (*gnmi.SubscribeRequest, error) {
@@ -26,12 +27,12 @@ func (c *Config) CreateDiffSubscribeRequest(cmd *cobra.Command) (*gnmi.Subscribe
 		Target:   c.DiffTarget,
 		Paths:    c.DiffPath,
 		Mode:     "ONCE",
-		Encoding: c.Encoding,
+		Encoding: &c.Encoding,
 	}
 	if flagIsSet(cmd, "qos") {
 		sc.Qos = &c.DiffQos
 	}
-	return c.CreateSubscribeRequest(sc, "")
+	return c.CreateSubscribeRequest(sc, nil)
 }
 
 func (c *Config) CreateDiffGetRequest() (*gnmi.GetRequest, error) {