diff --git a/.github/workflows/cloudevents-integration.yml b/.github/workflows/cloudevents-integration.yml index 5eae1f0eb..38951c85e 100644 --- a/.github/workflows/cloudevents-integration.yml +++ b/.github/workflows/cloudevents-integration.yml @@ -21,20 +21,6 @@ permissions: contents: read jobs: - # Disable mqtt-work-integration: because the current source client does not support increasing resource generation. - # mqtt-work-integration: - # name: mqtt-work-integration - # runs-on: ubuntu-latest - # steps: - # - name: checkout code - # uses: actions/checkout@v6.0.2 - # - name: install Go - # uses: actions/setup-go@v6 - # with: - # go-version: ${{ env.GO_VERSION }} - # - name: integration - # run: make test-cloudevents-work-mqtt-integration - grpc-work-integration: name: grpc-work-integration runs-on: ubuntu-latest diff --git a/go.mod b/go.mod index 9c5c35dc7..51293ec28 100644 --- a/go.mod +++ b/go.mod @@ -15,7 +15,6 @@ require ( github.com/google/cel-go v0.27.0 github.com/google/go-cmp v0.7.0 github.com/itchyny/gojq v0.12.18 - github.com/mochi-mqtt/server/v2 v2.7.9 github.com/onsi/ginkgo/v2 v2.28.1 github.com/onsi/gomega v1.39.1 github.com/openshift/api v0.0.0-20251125174858-5cf710f68a92 @@ -42,7 +41,7 @@ require ( k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 open-cluster-management.io/addon-framework v1.2.1-0.20260306083712-b6f9759b7b6d open-cluster-management.io/api v1.2.1-0.20260305152611-5bfebdbc3fdf - open-cluster-management.io/sdk-go v1.2.1-0.20260310072111-3041045c0177 + open-cluster-management.io/sdk-go v1.2.1-0.20260311085014-8db061d07611 sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03 sigs.k8s.io/cluster-inventory-api v0.0.0-20251124125836-445319b6307a sigs.k8s.io/controller-runtime v0.23.3 @@ -126,6 +125,7 @@ require ( github.com/mailru/easyjson v0.9.0 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect + github.com/mochi-mqtt/server/v2 v2.7.9 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect diff --git a/go.sum b/go.sum index 0b8e6f27d..ee4d0c25c 100644 --- a/go.sum +++ b/go.sum @@ -589,8 +589,8 @@ open-cluster-management.io/addon-framework v1.2.1-0.20260306083712-b6f9759b7b6d open-cluster-management.io/addon-framework v1.2.1-0.20260306083712-b6f9759b7b6d/go.mod h1:Bpw37w4GANroADMDR3F/ZUFoEuTKV9GIn4ijwICcK6E= open-cluster-management.io/api v1.2.1-0.20260305152611-5bfebdbc3fdf h1:SnLaZD2QHz+Ep2SfVKx9y5WIMmyLnQcEv/ySW8k/NXc= open-cluster-management.io/api v1.2.1-0.20260305152611-5bfebdbc3fdf/go.mod h1:ZpXs1bFTIIqKstMHdLO9IY0NFlbCvZgEtByvvNSmab0= -open-cluster-management.io/sdk-go v1.2.1-0.20260310072111-3041045c0177 h1:8YzKbl+PuWIvTyU7C1pVuPUIY4cdbccv5BXZySQfDYI= -open-cluster-management.io/sdk-go v1.2.1-0.20260310072111-3041045c0177/go.mod h1:lDef+5BvifXww0S7cseux+Wi8melkH29bAf33OZ0ZVg= +open-cluster-management.io/sdk-go v1.2.1-0.20260311085014-8db061d07611 h1:CNv03thePJrJNaxj5kUYT2f+pWmRf+HTX2UdWnCJaXk= +open-cluster-management.io/sdk-go v1.2.1-0.20260311085014-8db061d07611/go.mod h1:lDef+5BvifXww0S7cseux+Wi8melkH29bAf33OZ0ZVg= sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03 h1:1ShFiMjGQOR/8jTBkmZrk1gORxnvMwm1nOy2/DbHg4U= sigs.k8s.io/about-api v0.0.0-20250131010323-518069c31c03/go.mod h1:F1pT4mK53U6F16/zuaPSYpBaR7x5Kjym6aKJJC0/DHU= sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 h1:jpcvIRr3GLoUoEKRkHKSmGjxb6lWwrBlJsXc+eUYQHM= diff --git a/test/integration-test.mk b/test/integration-test.mk index ed8ce5ae5..f009070e6 100644 --- a/test/integration-test.mk +++ b/test/integration-test.mk @@ -38,19 +38,6 @@ test-addon-integration: envtest-setup ./addon-integration.test -ginkgo.slow-spec-threshold=15s -ginkgo.v -ginkgo.fail-fast ${ARGS} .PHONY: test-addon-integration -# In the cloud events scenario, skip the following tests -# - unmanaged_appliedwork_test.go, this test mainly focus on switching the hub kube-apiserver -# - manifestworkreplicaset_test.go, this test needs to update the work status with the hub work client, -# cloud events work client does not support it. (TODO) may add e2e to for mwrs. -test-cloudevents-work-mqtt-integration: envtest-setup build-work-integration - ./work-integration.test -ginkgo.slow-spec-threshold=15s -ginkgo.v -ginkgo.fail-fast \ - -ginkgo.skip-file manifestworkreplicaset_test.go \ - -ginkgo.skip-file unmanaged_appliedwork_test.go \ - -ginkgo.skip-file manifestworkgarbagecollection_test.go \ - -test.driver=mqtt \ - -v=4 ${ARGS} -.PHONY: test-cloudevents-work-mqtt-integration - # In the cloud events scenario, skip the following tests test-cloudevents-work-grpc-integration: envtest-setup build-work-integration ./work-integration.test -ginkgo.slow-spec-threshold=15s -ginkgo.v -ginkgo.fail-fast \ @@ -60,6 +47,3 @@ test-cloudevents-work-grpc-integration: envtest-setup build-work-integration test-integration: test-registration-operator-integration test-registration-integration test-placement-integration test-work-integration test-addon-integration .PHONY: test-integration - -test-cloudevents-integration: test-cloudevents-work-mqtt-integration test-cloudevents-work-grpc-integration -.PHONY: test-cloudevents-integration diff --git a/test/integration/util/mqtt.go b/test/integration/util/mqtt.go deleted file mode 100644 index 145b838e8..000000000 --- a/test/integration/util/mqtt.go +++ /dev/null @@ -1,92 +0,0 @@ -package util - -import ( - "fmt" - "log" - "os" - "time" - - mochimqtt "github.com/mochi-mqtt/server/v2" - "github.com/mochi-mqtt/server/v2/hooks/auth" - "github.com/mochi-mqtt/server/v2/listeners" - "gopkg.in/yaml.v2" - - "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt" - "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types" -) - -const MQTTBrokerHost = "127.0.0.1:1883" - -var mqttBroker *mochimqtt.Server - -func RunMQTTBroker() error { - // start a MQTT broker - mqttBroker = mochimqtt.New(nil) - - // allow all connections - if err := mqttBroker.AddHook(new(auth.AllowHook), nil); err != nil { - return err - } - - if err := mqttBroker.AddListener(listeners.NewTCP( - listeners.Config{ - ID: "mqtt-test-broker", - Address: MQTTBrokerHost, - })); err != nil { - return err - } - - go func() { - if err := mqttBroker.Serve(); err != nil { - log.Fatal(err) - } - }() - - return nil -} - -func StopMQTTBroker() error { - if mqttBroker != nil { - return mqttBroker.Close() - } - - return nil -} - -func CreateMQTTConfigFile(configFileName, sourceID string) error { - config := mqtt.MQTTConfig{ - BrokerHost: MQTTBrokerHost, - Topics: &types.Topics{ - SourceEvents: fmt.Sprintf("sources/%s/clusters/+/sourceevents", sourceID), - AgentEvents: fmt.Sprintf("sources/%s/clusters/+/agentevents", sourceID), - }, - } - - configData, err := yaml.Marshal(config) - if err != nil { - return err - } - - if err := os.WriteFile(configFileName, configData, 0600); err != nil { - return err - } - - return nil -} - -func NewMQTTSourceOptions(sourceID string) *mqtt.MQTTOptions { - return &mqtt.MQTTOptions{ - KeepAlive: 60, - PubQoS: 1, - SubQoS: 1, - Topics: types.Topics{ - SourceEvents: fmt.Sprintf("sources/%s/clusters/+/sourceevents", sourceID), - AgentEvents: fmt.Sprintf("sources/%s/clusters/+/agentevents", sourceID), - SourceBroadcast: "sources/+/sourcebroadcast", - }, - Dialer: &mqtt.MQTTDialer{ - BrokerHost: MQTTBrokerHost, - Timeout: 5 * time.Second, - }, - } -} diff --git a/test/integration/work/suite_test.go b/test/integration/work/suite_test.go index 031ca818d..abe50cb7d 100644 --- a/test/integration/work/suite_test.go +++ b/test/integration/work/suite_test.go @@ -14,7 +14,6 @@ import ( "github.com/onsi/gomega" "github.com/openshift/library-go/pkg/controller/controllercmd" "google.golang.org/grpc" - "k8s.io/apimachinery/pkg/util/rand" "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/rest" @@ -27,11 +26,7 @@ import ( workclientset "open-cluster-management.io/api/client/work/clientset/versioned" ocmfeature "open-cluster-management.io/api/feature" workapiv1 "open-cluster-management.io/api/work/v1" - "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/options" - "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work" "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload" - sourcecodec "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/source/codec" - workstore "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/store" pbv1 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/v1" cloudeventsgrpc "open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc" cemetrics "open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/metrics" @@ -150,31 +145,6 @@ var _ = ginkgo.BeforeSuite(func() { }) gomega.Expect(err).NotTo(gomega.HaveOccurred()) }() - case util.MQTTDriver: - sourceID := "work-test-mqtt" - err = util.RunMQTTBroker() - gomega.Expect(err).NotTo(gomega.HaveOccurred()) - - sourceConfigFileName = path.Join(tempDir, "mqttconfig") - err = util.CreateMQTTConfigFile(sourceConfigFileName, sourceID) - gomega.Expect(err).NotTo(gomega.HaveOccurred()) - - hubHash = helper.HubHash(util.MQTTBrokerHost) - - watcherStore, err := workstore.NewSourceLocalWatcherStore(envCtx, func(ctx context.Context) ([]*workapiv1.ManifestWork, error) { - return []*workapiv1.ManifestWork{}, nil - }) - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - clientOptions := options.NewGenericClientOptions( - util.NewMQTTSourceOptions(sourceID), - sourcecodec.NewManifestBundleCodec(), - fmt.Sprintf("%s-%s", sourceID, rand.String(5)), - ).WithSourceID(sourceID).WithClientWatcherStore(watcherStore) - sourceClient, err := work.NewSourceClientHolder(envCtx, clientOptions) - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - hubWorkClient = sourceClient.WorkInterface() case util.GRPCDriver: hubWorkClient, err = workclientset.NewForConfig(cfg) gomega.Expect(err).NotTo(gomega.HaveOccurred()) @@ -223,9 +193,6 @@ var _ = ginkgo.AfterSuite(func() { err := testEnv.Stop() gomega.Expect(err).ToNot(gomega.HaveOccurred()) - err = util.StopMQTTBroker() - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - if tempDir != "" { os.RemoveAll(tempDir) } diff --git a/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/allow_all.go b/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/allow_all.go deleted file mode 100644 index e05a0de3b..000000000 --- a/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/allow_all.go +++ /dev/null @@ -1,41 +0,0 @@ -// SPDX-License-Identifier: MIT -// SPDX-FileCopyrightText: 2022 mochi-mqtt, mochi-co -// SPDX-FileContributor: mochi-co - -package auth - -import ( - "bytes" - - "github.com/mochi-mqtt/server/v2" - "github.com/mochi-mqtt/server/v2/packets" -) - -// AllowHook is an authentication hook which allows connection access -// for all users and read and write access to all topics. -type AllowHook struct { - mqtt.HookBase -} - -// ID returns the ID of the hook. -func (h *AllowHook) ID() string { - return "allow-all-auth" -} - -// Provides indicates which hook methods this hook provides. -func (h *AllowHook) Provides(b byte) bool { - return bytes.Contains([]byte{ - mqtt.OnConnectAuthenticate, - mqtt.OnACLCheck, - }, []byte{b}) -} - -// OnConnectAuthenticate returns true/allowed for all requests. -func (h *AllowHook) OnConnectAuthenticate(cl *mqtt.Client, pk packets.Packet) bool { - return true -} - -// OnACLCheck returns true/allowed for all checks. -func (h *AllowHook) OnACLCheck(cl *mqtt.Client, topic string, write bool) bool { - return true -} diff --git a/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/auth.go b/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/auth.go deleted file mode 100644 index d50cc5cee..000000000 --- a/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/auth.go +++ /dev/null @@ -1,103 +0,0 @@ -// SPDX-License-Identifier: MIT -// SPDX-FileCopyrightText: 2022 mochi-mqtt, mochi-co -// SPDX-FileContributor: mochi-co - -package auth - -import ( - "bytes" - - mqtt "github.com/mochi-mqtt/server/v2" - "github.com/mochi-mqtt/server/v2/packets" -) - -// Options contains the configuration/rules data for the auth ledger. -type Options struct { - Data []byte - Ledger *Ledger -} - -// Hook is an authentication hook which implements an auth ledger. -type Hook struct { - mqtt.HookBase - config *Options - ledger *Ledger -} - -// ID returns the ID of the hook. -func (h *Hook) ID() string { - return "auth-ledger" -} - -// Provides indicates which hook methods this hook provides. -func (h *Hook) Provides(b byte) bool { - return bytes.Contains([]byte{ - mqtt.OnConnectAuthenticate, - mqtt.OnACLCheck, - }, []byte{b}) -} - -// Init configures the hook with the auth ledger to be used for checking. -func (h *Hook) Init(config any) error { - if _, ok := config.(*Options); !ok && config != nil { - return mqtt.ErrInvalidConfigType - } - - if config == nil { - config = new(Options) - } - - h.config = config.(*Options) - - var err error - if h.config.Ledger != nil { - h.ledger = h.config.Ledger - } else if len(h.config.Data) > 0 { - h.ledger = new(Ledger) - err = h.ledger.Unmarshal(h.config.Data) - } - if err != nil { - return err - } - - if h.ledger == nil { - h.ledger = &Ledger{ - Auth: AuthRules{}, - ACL: ACLRules{}, - } - } - - h.Log.Info("loaded auth rules", - "authentication", len(h.ledger.Auth), - "acl", len(h.ledger.ACL)) - - return nil -} - -// OnConnectAuthenticate returns true if the connecting client has rules which provide access -// in the auth ledger. -func (h *Hook) OnConnectAuthenticate(cl *mqtt.Client, pk packets.Packet) bool { - if _, ok := h.ledger.AuthOk(cl, pk); ok { - return true - } - - h.Log.Info("client failed authentication check", - "username", string(pk.Connect.Username), - "remote", cl.Net.Remote) - return false -} - -// OnACLCheck returns true if the connecting client has matching read or write access to subscribe -// or publish to a given topic. -func (h *Hook) OnACLCheck(cl *mqtt.Client, topic string, write bool) bool { - if _, ok := h.ledger.ACLOk(cl, topic, write); ok { - return true - } - - h.Log.Debug("client failed allowed ACL check", - "client", cl.ID, - "username", string(cl.Properties.Username), - "topic", topic) - - return false -} diff --git a/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/ledger.go b/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/ledger.go deleted file mode 100644 index 694b19da1..000000000 --- a/vendor/github.com/mochi-mqtt/server/v2/hooks/auth/ledger.go +++ /dev/null @@ -1,246 +0,0 @@ -// SPDX-License-Identifier: MIT -// SPDX-FileCopyrightText: 2022 mochi-mqtt, mochi-co -// SPDX-FileContributor: mochi-co - -package auth - -import ( - "encoding/json" - "strings" - "sync" - - "gopkg.in/yaml.v3" - - "github.com/mochi-mqtt/server/v2" - "github.com/mochi-mqtt/server/v2/packets" -) - -const ( - Deny Access = iota // user cannot access the topic - ReadOnly // user can only subscribe to the topic - WriteOnly // user can only publish to the topic - ReadWrite // user can both publish and subscribe to the topic -) - -// Access determines the read/write privileges for an ACL rule. -type Access byte - -// Users contains a map of access rules for specific users, keyed on username. -type Users map[string]UserRule - -// UserRule defines a set of access rules for a specific user. -type UserRule struct { - Username RString `json:"username,omitempty" yaml:"username,omitempty"` // the username of a user - Password RString `json:"password,omitempty" yaml:"password,omitempty"` // the password of a user - ACL Filters `json:"acl,omitempty" yaml:"acl,omitempty"` // filters to match, if desired - Disallow bool `json:"disallow,omitempty" yaml:"disallow,omitempty"` // allow or disallow the user -} - -// AuthRules defines generic access rules applicable to all users. -type AuthRules []AuthRule - -type AuthRule struct { - Client RString `json:"client,omitempty" yaml:"client,omitempty"` // the id of a connecting client - Username RString `json:"username,omitempty" yaml:"username,omitempty"` // the username of a user - Remote RString `json:"remote,omitempty" yaml:"remote,omitempty"` // remote address or - Password RString `json:"password,omitempty" yaml:"password,omitempty"` // the password of a user - Allow bool `json:"allow,omitempty" yaml:"allow,omitempty"` // allow or disallow the users -} - -// ACLRules defines generic topic or filter access rules applicable to all users. -type ACLRules []ACLRule - -// ACLRule defines access rules for a specific topic or filter. -type ACLRule struct { - Client RString `json:"client,omitempty" yaml:"client,omitempty"` // the id of a connecting client - Username RString `json:"username,omitempty" yaml:"username,omitempty"` // the username of a user - Remote RString `json:"remote,omitempty" yaml:"remote,omitempty"` // remote address or - Filters Filters `json:"filters,omitempty" yaml:"filters,omitempty"` // filters to match -} - -// Filters is a map of Access rules keyed on filter. -type Filters map[RString]Access - -// RString is a rule value string. -type RString string - -// Matches returns true if the rule matches a given string. -func (r RString) Matches(a string) bool { - rr := string(r) - if r == "" || r == "*" || a == rr { - return true - } - - i := strings.Index(rr, "*") - if i > 0 && len(a) > i && strings.Compare(rr[:i], a[:i]) == 0 { - return true - } - - return false -} - -// FilterMatches returns true if a filter matches a topic rule. -func (r RString) FilterMatches(a string) bool { - _, ok := MatchTopic(string(r), a) - return ok -} - -// MatchTopic checks if a given topic matches a filter, accounting for filter -// wildcards. Eg. filter /a/b/+/c == topic a/b/d/c. -func MatchTopic(filter string, topic string) (elements []string, matched bool) { - filterParts := strings.Split(filter, "/") - topicParts := strings.Split(topic, "/") - - elements = make([]string, 0) - for i := 0; i < len(filterParts); i++ { - if i >= len(topicParts) { - matched = false - return - } - - if filterParts[i] == "+" { - elements = append(elements, topicParts[i]) - continue - } - - if filterParts[i] == "#" { - matched = true - elements = append(elements, strings.Join(topicParts[i:], "/")) - return - } - - if filterParts[i] != topicParts[i] { - matched = false - return - } - } - - return elements, true -} - -// Ledger is an auth ledger containing access rules for users and topics. -type Ledger struct { - sync.Mutex `json:"-" yaml:"-"` - Users Users `json:"users" yaml:"users"` - Auth AuthRules `json:"auth" yaml:"auth"` - ACL ACLRules `json:"acl" yaml:"acl"` -} - -// Update updates the internal values of the ledger. -func (l *Ledger) Update(ln *Ledger) { - l.Lock() - defer l.Unlock() - l.Auth = ln.Auth - l.ACL = ln.ACL -} - -// AuthOk returns true if the rules indicate the user is allowed to authenticate. -func (l *Ledger) AuthOk(cl *mqtt.Client, pk packets.Packet) (n int, ok bool) { - // If the users map is set, always check for a predefined user first instead - // of iterating through global rules. - if l.Users != nil { - if u, ok := l.Users[string(cl.Properties.Username)]; ok && - u.Password != "" && - u.Password == RString(pk.Connect.Password) { - return 0, !u.Disallow - } - } - - // If there's no users map, or no user was found, attempt to find a matching - // rule (which may also contain a user). - for n, rule := range l.Auth { - if rule.Client.Matches(cl.ID) && - rule.Username.Matches(string(cl.Properties.Username)) && - rule.Password.Matches(string(pk.Connect.Password)) && - rule.Remote.Matches(cl.Net.Remote) { - return n, rule.Allow - } - } - - return 0, false -} - -// ACLOk returns true if the rules indicate the user is allowed to read or write to -// a specific filter or topic respectively, based on the `write` bool. -func (l *Ledger) ACLOk(cl *mqtt.Client, topic string, write bool) (n int, ok bool) { - // If the users map is set, always check for a predefined user first instead - // of iterating through global rules. - if l.Users != nil { - if u, ok := l.Users[string(cl.Properties.Username)]; ok && len(u.ACL) > 0 { - for filter, access := range u.ACL { - if filter.FilterMatches(topic) { - if !write && (access == ReadOnly || access == ReadWrite) { - return n, true - } else if write && (access == WriteOnly || access == ReadWrite) { - return n, true - } else { - return n, false - } - } - } - } - } - - for n, rule := range l.ACL { - if rule.Client.Matches(cl.ID) && - rule.Username.Matches(string(cl.Properties.Username)) && - rule.Remote.Matches(cl.Net.Remote) { - if len(rule.Filters) == 0 { - return n, true - } - - if write { - for filter, access := range rule.Filters { - if access == WriteOnly || access == ReadWrite { - if filter.FilterMatches(topic) { - return n, true - } - } - } - } - - if !write { - for filter, access := range rule.Filters { - if access == ReadOnly || access == ReadWrite { - if filter.FilterMatches(topic) { - return n, true - } - } - } - } - - for filter := range rule.Filters { - if filter.FilterMatches(topic) { - return n, false - } - } - } - } - - return 0, true -} - -// ToJSON encodes the values into a JSON string. -func (l *Ledger) ToJSON() (data []byte, err error) { - return json.Marshal(l) -} - -// ToYAML encodes the values into a YAML string. -func (l *Ledger) ToYAML() (data []byte, err error) { - return yaml.Marshal(l) -} - -// Unmarshal decodes a JSON or YAML string (such as a rule config from a file) into a struct. -func (l *Ledger) Unmarshal(data []byte) error { - l.Lock() - defer l.Unlock() - if len(data) == 0 { - return nil - } - - if data[0] == '{' { - return json.Unmarshal(data, l) - } - - return yaml.Unmarshal(data, &l) -} diff --git a/vendor/modules.txt b/vendor/modules.txt index 0df7139fb..32aa797ad 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -455,7 +455,6 @@ github.com/mitchellh/reflectwalk # github.com/mochi-mqtt/server/v2 v2.7.9 ## explicit; go 1.21 github.com/mochi-mqtt/server/v2 -github.com/mochi-mqtt/server/v2/hooks/auth github.com/mochi-mqtt/server/v2/hooks/storage github.com/mochi-mqtt/server/v2/listeners github.com/mochi-mqtt/server/v2/mempool @@ -1973,7 +1972,7 @@ open-cluster-management.io/api/operator/v1 open-cluster-management.io/api/utils/work/v1/workapplier open-cluster-management.io/api/work/v1 open-cluster-management.io/api/work/v1alpha1 -# open-cluster-management.io/sdk-go v1.2.1-0.20260310072111-3041045c0177 +# open-cluster-management.io/sdk-go v1.2.1-0.20260311085014-8db061d07611 ## explicit; go 1.25.0 open-cluster-management.io/sdk-go/pkg/apis/cluster/v1alpha1 open-cluster-management.io/sdk-go/pkg/apis/cluster/v1beta1 diff --git a/vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/store/local.go b/vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/store/local.go deleted file mode 100644 index afed44366..000000000 --- a/vendor/open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/store/local.go +++ /dev/null @@ -1,256 +0,0 @@ -package store - -import ( - "context" - "errors" - "fmt" - "time" - - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/util/wait" - "k8s.io/apimachinery/pkg/watch" - "k8s.io/client-go/tools/cache" - "k8s.io/client-go/util/workqueue" - "k8s.io/klog/v2" - - workv1 "open-cluster-management.io/api/work/v1" - - "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/store" - "open-cluster-management.io/sdk-go/pkg/cloudevents/clients/utils" -) - -// ListLocalWorksFunc loads the works from the local environment. -type ListLocalWorksFunc func(ctx context.Context) ([]*workv1.ManifestWork, error) - -type watchEvent struct { - Key string - Type watch.EventType -} - -var _ store.ClientWatcherStore[*workv1.ManifestWork] = &SourceLocalWatcherStore{} - -// SourceLocalWatcherStore caches the works in this local store and provide the watch ability by watch event channel. -// -// It is used for building ManifestWork source client. -type SourceLocalWatcherStore struct { - baseSourceStore - watcher *store.Watcher - eventQueue cache.Queue -} - -// NewSourceLocalWatcherStore returns a LocalWatcherStore with works that list by ListLocalWorksFunc -func NewSourceLocalWatcherStore(ctx context.Context, listFunc ListLocalWorksFunc) (*SourceLocalWatcherStore, error) { - works, err := listFunc(ctx) - if err != nil { - return nil, err - } - - // A local localStore to cache the works - localStore := cache.NewStore(cache.MetaNamespaceKeyFunc) - for _, work := range works { - if errs := utils.ValidateWork(work); len(errs) != 0 { - return nil, fmt.Errorf("%s", errs.ToAggregate().Error()) - } - - if err := localStore.Add(work.DeepCopy()); err != nil { - return nil, err - } - } - - s := &SourceLocalWatcherStore{ - baseSourceStore: baseSourceStore{ - BaseClientWatchStore: store.BaseClientWatchStore[*workv1.ManifestWork]{ - Store: localStore, - Initiated: true, - }, - - // A queue to save the received work events, it helps us retry events - // where errors occurred while processing - receivedWorks: workqueue.NewTypedRateLimitingQueueWithConfig( - workqueue.DefaultTypedControllerRateLimiter[*workv1.ManifestWork](), - workqueue.TypedRateLimitingQueueConfig[*workv1.ManifestWork]{Name: "local-watcher-store"}, - ), - }, - - watcher: store.NewWatcher(), - - // A queue to save the work client send events, if run a client without a watcher, - // it will block the client, this queue helps to resolve this blocking. - // Only save the latest event for a work. - eventQueue: cache.NewFIFO(func(obj interface{}) (string, error) { - evt, ok := obj.(*watchEvent) - if !ok { - return "", fmt.Errorf("unknown object type %T", obj) - } - - return evt.Key, nil - }), - } - - // start a goroutine to process the received work events from the work queue with current store. - go newWorkProcessor(s.receivedWorks, s).run(ctx) - - // start a goroutine to handle the events that are produced by work client - go wait.UntilWithContext(ctx, s.processLoop, time.Second) - - return s, nil -} - -// Add a work to the cache and send an event to the event queue -func (s *SourceLocalWatcherStore) Add(work runtime.Object) error { - s.Lock() - defer s.Unlock() - - if err := s.Store.Add(work); err != nil { - return err - } - - key, err := key(work) - if err != nil { - return err - } - - return s.eventQueue.Add(&watchEvent{Key: key, Type: watch.Added}) -} - -// Update a work in the cache and send an event to the event queue -func (s *SourceLocalWatcherStore) Update(work runtime.Object) error { - s.Lock() - defer s.Unlock() - - if err := s.Store.Update(work); err != nil { - return err - } - - key, err := key(work) - if err != nil { - return err - } - - return s.eventQueue.Update(&watchEvent{Key: key, Type: watch.Modified}) -} - -// Delete a work from the cache and send an event to the event queue -func (s *SourceLocalWatcherStore) Delete(work runtime.Object) error { - s.Lock() - defer s.Unlock() - - if err := s.Store.Delete(work); err != nil { - return err - } - - key, err := key(work) - if err != nil { - return err - } - - return s.eventQueue.Update(&watchEvent{Key: key, Type: watch.Deleted}) -} - -func (s *SourceLocalWatcherStore) HasInitiated() bool { - return s.Initiated -} - -func (s *SourceLocalWatcherStore) GetWatcher(ctx context.Context, namespace string, opts metav1.ListOptions) (watch.Interface, error) { - // TODO may consider to support watch with namespace - if namespace != metav1.NamespaceAll { - return nil, fmt.Errorf("unsupported to watch from the namespace %s", namespace) - } - - // If AllowWatchBookmarks is enabled, send a bookmark event to signal the end of the initial event stream. - // This is required by Kubernetes 1.35+ reflectors to properly initialize watches. - if opts.AllowWatchBookmarks { - // Send the bookmark event asynchronously to avoid blocking the watch initialization - go func() { - // Create a minimal work object for the bookmark event with the required annotation - bookmarkWork := &workv1.ManifestWork{} - bookmarkWork.SetResourceVersion(opts.ResourceVersion) - bookmarkWork.SetAnnotations(map[string]string{ - metav1.InitialEventsAnnotationKey: "true", - }) - s.watcher.Receive(watch.Event{Type: watch.Bookmark, Object: bookmarkWork}) - }() - } - - return s.watcher, nil -} - -// processLoop drains the work event queue and send the event to the watch channel. -func (s *SourceLocalWatcherStore) processLoop(ctx context.Context) { - logger := klog.FromContext(ctx) - for { - // this will be blocked until the event queue has events - obj, err := s.eventQueue.Pop(func(interface{}, bool) error { - // do nothing - return nil - }) - if err != nil { - if err == cache.ErrFIFOClosed { - return - } - - logger.Error(err, "failed to pop the object, requeue it", "object", obj) - // this is the safe way to re-enqueue. - if err := s.eventQueue.Add(obj); err != nil { - logger.Error(err, "failed to requeue the obj", "object", obj) - return - } - } - - evt, ok := obj.(*watchEvent) - if !ok { - logger.Error(errors.New("unknown the object type from the event queue"), "", "object", obj) - return - } - - obj, exists, err := s.Store.GetByKey(evt.Key) - if err != nil { - logger.Error(err, "failed to get the work", "key", evt.Key) - return - } - - if !exists { - if evt.Type == watch.Deleted { - namespace, name, err := cache.SplitMetaNamespaceKey(evt.Key) - if err != nil { - logger.Error(err, "unexpected event key", "key", evt.Key) - return - } - - // the work has been deleted, return a work only with its namespace and name - // this will be blocked until this event is consumed - s.watcher.Receive(watch.Event{ - Type: watch.Deleted, - Object: &workv1.ManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: name, - Namespace: namespace, - }, - }, - }) - return - } - - logger.Error(errors.New("the work does not exist in the cache"), "", "key", evt.Key) - return - } - - work, ok := obj.(*workv1.ManifestWork) - if !ok { - logger.Error(fmt.Errorf("unknown object type %T from the cache", obj), "") - return - } - - // this will be blocked until this event is consumed - s.watcher.Receive(watch.Event{Type: evt.Type, Object: work}) - } -} - -func key(obj runtime.Object) (string, error) { - work, ok := obj.(*workv1.ManifestWork) - if !ok { - return "", fmt.Errorf("obj %T is not a work", obj) - } - return work.Namespace + "/" + work.Name, nil -}