Merge pull request #669 from cgalibern/dev

Add support for om,ox node event --wait ...
opensvc · Jan 20, 2025 · bbb5413 · bbb5413
2 parents 7f43073 + 96fcb5d
commit bbb5413
Show file tree

Hide file tree

Showing 68 changed files with 1,176 additions and 312 deletions.
diff --git a/core/client/api/get_events.go b/core/client/api/get_events.go
@@ -24,6 +24,7 @@ type GetEvents struct {
 	relatives *bool
 	Limit     *uint64
 	Filters   []string
+	Wait      bool
 	Duration  *time.Duration
 }
 
@@ -42,6 +43,11 @@ func (t *GetEvents) SetFilters(filters []string) *GetEvents {
 	return t
 }
 
+func (t *GetEvents) SetWait(wait bool) *GetEvents {
+	t.Wait = wait
+	return t
+}
+
 func (t *GetEvents) SetNamespace(s string) *GetEvents {
 	t.namespace = &s
 	return t
@@ -179,6 +185,7 @@ func (t GetEvents) eventsBase() (*http.Response, error) {
 	params := api.GetDaemonEventsParams{
 		Filter:   &t.Filters,
 		Selector: t.selector,
+		Cache:    &t.Wait,
 	}
 	if t.Limit != nil {
 		i := int64(*t.Limit)

diff --git a/core/instance/data.go b/core/instance/data.go
@@ -78,8 +78,8 @@ func (c *Data[T]) DropNode(nodename string) {
 	delete(c.nodeToPath, nodename)
 }
 
-// Get returns an instance data or nil if data is not found
-func (c *Data[T]) Get(p naming.Path, nodename string) *T {
+// GetByPathAndNode returns an instance data or nil if data is not found
+func (c *Data[T]) GetByPathAndNode(p naming.Path, nodename string) *T {
 	c.RLock()
 	v := c.data[InstanceString(p, nodename)]
 	c.RUnlock()

diff --git a/core/instance/data_test.go b/core/instance/data_test.go
@@ -17,10 +17,10 @@ func Test_Mapper(t *testing.T) {
 	mapper.Set(p, "node2", &Status{Avail: status.Warn})
 	mapper.Set(p2, "node2", &Status{Avail: status.Down})
 
-	require.Equal(t, status.Up, mapper.Get(p, "node1").Avail)
-	require.Equal(t, status.Warn, mapper.Get(p, "node2").Avail)
-	require.Equal(t, status.Down, mapper.Get(p2, "node2").Avail)
-	require.Nil(t, mapper.Get(p, "node3"))
+	require.Equal(t, status.Up, mapper.GetByPathAndNode(p, "node1").Avail)
+	require.Equal(t, status.Warn, mapper.GetByPathAndNode(p, "node2").Avail)
+	require.Equal(t, status.Down, mapper.GetByPathAndNode(p2, "node2").Avail)
+	require.Nil(t, mapper.GetByPathAndNode(p, "node3"))
 	require.Len(t, mapper.GetAll(), 3)
 	require.Len(t, mapper.GetByPath(p), 2)
 	require.Len(t, mapper.GetByNode("node1"), 1)
@@ -29,9 +29,9 @@ func Test_Mapper(t *testing.T) {
 	mapper.Unset(p, "node3")
 	mapper.Unset(p, "node2")
 
-	require.Equal(t, status.Up, mapper.Get(p, "node1").Avail)
-	require.Nil(t, mapper.Get(p, "node2"))
-	require.Nil(t, mapper.Get(p, "node3"))
+	require.Equal(t, status.Up, mapper.GetByPathAndNode(p, "node1").Avail)
+	require.Nil(t, mapper.GetByPathAndNode(p, "node2"))
+	require.Nil(t, mapper.GetByPathAndNode(p, "node3"))
 	require.Len(t, mapper.GetAll(), 2)
 	require.Len(t, mapper.GetByPath(p), 1)
 	require.Len(t, mapper.GetByNode("node1"), 1)

diff --git a/core/node/data.go b/core/node/data.go
@@ -65,8 +65,8 @@ func (c *Data[T]) Unset(nodename string) {
 	delete(c.data, nodename)
 }
 
-// Get return the stored value for nodename or nil if not found
-func (c *Data[T]) Get(nodename string) *T {
+// GetByNode return the stored value for nodename or nil if not found
+func (c *Data[T]) GetByNode(nodename string) *T {
 	c.RLock()
 	v := c.data[nodename]
 	c.RUnlock()

diff --git a/core/object/data.go b/core/object/data.go
@@ -46,7 +46,7 @@ func (c *Data[T]) Unset(p naming.Path) {
 	c.Unlock()
 }
 
-func (c *Data[T]) Get(p naming.Path) *T {
+func (c *Data[T]) GetByPath(p naming.Path) *T {
 	c.RLock()
 	v := c.data[p]
 	c.RUnlock()

diff --git a/core/om/factory.go b/core/om/factory.go
@@ -5,7 +5,6 @@ import (
 	_ "embed"
 	"fmt"
 	"os"
-
 	"time"
 
 	"github.com/spf13/cobra"
@@ -1165,7 +1164,7 @@ func newCmdNodeEvents() *cobra.Command {
 	addFlagEventTemplate(flags, &options.Template)
 	addFlagWait(flags, &options.Wait)
 	addFlagNodeSelector(flags, &options.NodeSelector)
-	flags.Uint64Var(&options.Limit, "limit", 0, "limit event count to fetch")
+	flags.Uint64Var(&options.Limit, "limit", 0, "stop listening when <limit> events are received, the default is 0 (unlimited) or 1 if --wait is set")
 	return cmd
 }
 

diff --git a/core/omcmd/node_events.go b/core/omcmd/node_events.go
@@ -135,6 +135,9 @@ func hasInstanceLabel(labels []pubsub.Label, expected ...string) bool {
 }
 
 func (t *CmdNodeEvents) Run() error {
+	if t.Wait && t.Limit == 0 {
+		t.Limit = 1
+	}
 	if t.Local {
 		t.NodeSelector = hostname.Hostname()
 	}
@@ -320,6 +323,7 @@ func (t *CmdNodeEvents) getEvReader(nodename string) (event.ReadCloser, error) {
 	return t.cli.NewGetEvents().
 		SetRelatives(false).
 		SetLimit(t.Limit).
+		SetWait(t.Wait).
 		SetFilters(t.Filters).
 		SetDuration(t.Duration).
 		SetNodename(nodename).

diff --git a/core/ox/factory.go b/core/ox/factory.go
@@ -1262,7 +1262,7 @@ func newCmdNodeEvents() *cobra.Command {
 	addFlagEventTemplate(flags, &options.Template)
 	addFlagWait(flags, &options.Wait)
 	addFlagNodeSelector(flags, &options.NodeSelector)
-	flags.Uint64Var(&options.Limit, "limit", 0, "limit event count to fetch")
+	flags.Uint64Var(&options.Limit, "limit", 0, "stop listening when <limit> events are received, the default is 0 (unlimited) or 1 if --wait is set")
 	return cmd
 }
 

diff --git a/core/oxcmd/node_events.go b/core/oxcmd/node_events.go
@@ -2,14 +2,13 @@ package oxcmd
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"strings"
 	"text/template"
 	"time"
 
-	"encoding/json"
-
 	"github.com/opensvc/om3/core/client"
 	"github.com/opensvc/om3/core/clientcontext"
 	"github.com/opensvc/om3/core/event"
@@ -30,8 +29,9 @@ type (
 		Limit    uint64
 		Template string
 		Wait     bool
-		templ    *template.Template
-		helper   *templateHelper
+
+		templ  *template.Template
+		helper *templateHelper
 
 		cli          *client.T
 		NodeSelector string
@@ -136,6 +136,9 @@ func hasInstanceLabel(labels []pubsub.Label, expected ...string) bool {
 }
 
 func (t *CmdNodeEvents) Run() error {
+	if t.Wait && t.Limit == 0 {
+		t.Limit = 1
+	}
 	if !clientcontext.IsSet() && t.NodeSelector == "" {
 		t.NodeSelector = hostname.Hostname()
 	}
@@ -313,6 +316,7 @@ func (t *CmdNodeEvents) getEvReader(nodename string) (event.ReadCloser, error) {
 	return t.cli.NewGetEvents().
 		SetRelatives(false).
 		SetLimit(t.Limit).
+		SetWait(t.Wait).
 		SetFilters(t.Filters).
 		SetDuration(t.Duration).
 		SetNodename(nodename).

diff --git a/core/schedule/data.go b/core/schedule/data.go
@@ -46,7 +46,7 @@ func (c *Data[T]) Unset(p naming.Path) {
 	c.Unlock()
 }
 
-func (c *Data[T]) Get(p naming.Path) *T {
+func (c *Data[T]) GetByPath(p naming.Path) *T {
 	c.RLock()
 	v := c.data[p]
 	c.RUnlock()

diff --git a/daemon/api/api.yaml b/daemon/api/api.yaml
@@ -2,7 +2,7 @@ openapi: 3.0.0
 
 info:
   title: opensvc agent api
-  version: 3.12.2
+  version: 3.12.3
 
 paths:
   /auth/info:
@@ -405,6 +405,7 @@ paths:
         - $ref: '#/components/parameters/Duration'
         - $ref: '#/components/parameters/Limit'
         - $ref: '#/components/parameters/EventFilter'
+        - $ref: '#/components/parameters/EventCache'
         - $ref: '#/components/parameters/SelectorOptional'
       responses:
         200:
@@ -6841,6 +6842,13 @@ components:
       description: report this number of past last log entries
       schema:
         type: integer
+    EventCache:
+      name: cache
+      in: query
+      description: start the event feed using the latest past events from the cache that match the specified kind and labels.
+      schema:
+        type: boolean
+        default: false
     EventFilter:
       name: filter
       in: query
@@ -6850,7 +6858,7 @@ components:
         items:
           type: string
           description: |
-            filter expression: [kind][,label=value]*
+            filter expression: [kind][,label=value]*[.dataxxx=value]
           example: ObjectStatusUpdated,path=foo
     DRBDConfigName:
       name: name

diff --git a/daemon/api/codegen_client_gen.go b/daemon/api/codegen_client_gen.go