Auto reconnect if nats credential expires

Signed-off-by: Tamal Saha <tamal@appscode.com>

Author: tamalsaha

go.mod

@@ -8,10 +8,9 @@ require (
 	github.com/cloudevents/sdk-go/v2 v2.15.2
 	github.com/nats-io/nats.go v1.38.0
 	github.com/pkg/errors v0.9.1
-	go.bytebuilders.dev/license-verifier v0.14.4
-	go.bytebuilders.dev/license-verifier/kubernetes v0.14.4
+	go.bytebuilders.dev/license-verifier v0.14.6
+	go.bytebuilders.dev/license-verifier/kubernetes v0.14.6
 	gomodules.xyz/counter v0.0.1
-	gomodules.xyz/sync v0.1.0
 	k8s.io/api v0.30.2
 	k8s.io/apimachinery v0.30.2
 	k8s.io/client-go v0.30.2
@@ -91,7 +90,7 @@ require (
 	github.com/yudai/gojsondiff v1.0.0 // indirect
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
-	go.bytebuilders.dev/license-proxyserver v0.0.19 // indirect
+	go.bytebuilders.dev/license-proxyserver v0.0.20 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.31.0 // indirect

go.sum

@@ -210,12 +210,12 @@ github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
 github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
 github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
 github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
-go.bytebuilders.dev/license-proxyserver v0.0.19 h1:mY7zPDN0JCw2a1UajOuQUQKQKjjm5KBx2CbkT/+a1N8=
-go.bytebuilders.dev/license-proxyserver v0.0.19/go.mod h1:B3Ig2Fo1qUollSV9GgfyFK8tXBI0RmUSpP1KFMZ2N7Q=
-go.bytebuilders.dev/license-verifier v0.14.4 h1:JwTGQFew4nudwv8Pk3BdfQRts8KfgUQ5xhu138w1wt4=
-go.bytebuilders.dev/license-verifier v0.14.4/go.mod h1:LqWXJKee5ofDcCYM6T5WilYlUc4NlKeZz58tHwO8GEs=
-go.bytebuilders.dev/license-verifier/kubernetes v0.14.4 h1:NeHq6SuVhRIVaMW2kSXdr8DcuUOg2jVL9rsODIQp9Fc=
-go.bytebuilders.dev/license-verifier/kubernetes v0.14.4/go.mod h1:1C7SaOJShC60mIXP1hXBaDWGpb0hrHQ4p4nUEvI6YQY=
+go.bytebuilders.dev/license-proxyserver v0.0.20 h1:gzRSwUmX/LSwPVE6T9oy5RLIutU1EeI7hmS+QGsYBY4=
+go.bytebuilders.dev/license-proxyserver v0.0.20/go.mod h1:2PJmjMCXncVyeP3fIVQ+hwZnuhmWSTmbcuEMBrFKIac=
+go.bytebuilders.dev/license-verifier v0.14.6 h1:0iHYGURUbx8toiXvFKftn/qMpeHzqHbAgEnEzOCNLvo=
+go.bytebuilders.dev/license-verifier v0.14.6/go.mod h1:LqWXJKee5ofDcCYM6T5WilYlUc4NlKeZz58tHwO8GEs=
+go.bytebuilders.dev/license-verifier/kubernetes v0.14.6 h1:NxmASX0A3lu+ABd4zuT5Ib+I63y3j5uJxmlUFEGxqWg=
+go.bytebuilders.dev/license-verifier/kubernetes v0.14.6/go.mod h1:N5QxsJF4EGLduOsTsW9gGfRuuMvN33T8pg5Y9NfKzuo=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
@@ -283,8 +283,6 @@ gomodules.xyz/mergo v0.3.13 h1:q6cL/MMXZH/MrR2+yjSihFFq6UifXqjwaqI48B6cMEM=
 gomodules.xyz/mergo v0.3.13/go.mod h1:F/2rKC7j0URTnHUKDiTiLcGdLMhdv8jK2Za3cRTUVmc=
 gomodules.xyz/pointer v0.1.0 h1:sG2UKrYVSo6E3r4itAjXfPfe4fuXMi0KdyTHpR3vGCg=
 gomodules.xyz/pointer v0.1.0/go.mod h1:sPLsC0+yLTRecUiC5yVlyvXhZ6LAGojNCRWNNqoplvo=
-gomodules.xyz/sync v0.1.0 h1:Y7vHOtMrqN9FojRwkCQdC17dKL1fVx+6xb7WdfnXX58=
-gomodules.xyz/sync v0.1.0/go.mod h1:kv570yCdknyiZ8Y94uaRFGBC5E47TV/5A7PD9jlnJoQ=
 gomodules.xyz/testing v0.0.4 h1:XGKt4B64mBe7P9kPR0Rz1nCQpWoSpBEFdTGkfU1RLe4=
 gomodules.xyz/testing v0.0.4/go.mod h1:hD6aXtv9eVycPwS01zv+QTl5BrK2DXQgr6bHqnrW+44=
 gomodules.xyz/x v0.0.17 h1:Ik3wf0suCMiYPY0miFUh+q8BpjsUHc/7zvANbFViBQA=

lib/nats.go

@@ -24,6 +24,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -45,10 +46,22 @@ const (
 )
 
 type NatsConfig struct {
-	// LicenseID string     `json:"licenseID"`
-	Subject string     `json:"natsSubject"`
-	Server  string     `json:"natsServer"`
-	Client  *nats.Conn `json:"-"`
+	Subject string `json:"natsSubject"`
+	Server  string `json:"natsServer"`
+}
+
+type NatsClient struct {
+	cfg         *rest.Config
+	clusterID   string
+	LicenseFile string
+
+	le *kubernetes.LicenseEnforcer
+	l  *v1alpha1.License
+
+	nc      *nats.Conn
+	Subject string
+	Server  string
+	mu      sync.Mutex
 }
 
 // NatsCredential represents the api response of the register licensed user api
@@ -61,75 +74,118 @@ type LicenseIDGetter interface {
 	GetLicenseID() string
 }
 
-type LicenseUpdater struct {
-	le      *kubernetes.LicenseEnforcer
-	License v1alpha1.License
-	mu      sync.Mutex
+func (c *NatsClient) Request(data []byte, timeout time.Duration) (*nats.Msg, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	var justConnected bool
+	if c.nc == nil {
+		if err := c.connect(); err != nil {
+			return nil, err
+		}
+		justConnected = true
+	}
+	msg, err := c.nc.Request(c.Subject, data, timeout)
+	if err != nil && !justConnected && isNatsAuthError(err.Error()) {
+		if err := c.connect(); err != nil {
+			return nil, err
+		}
+		msg, err = c.nc.Request(c.Subject, data, timeout)
+	}
+	return msg, err
+}
+
+// src: https://github.com/nats-io/nats.go/blob/main/nats.go#L3693-L3709
+func isNatsAuthError(e string) bool {
+	if strings.HasPrefix(e, nats.AUTHORIZATION_ERR) {
+		return true
+	}
+	if strings.HasPrefix(e, nats.AUTHENTICATION_EXPIRED_ERR) {
+		return true
+	}
+	if strings.HasPrefix(e, nats.AUTHENTICATION_REVOKED_ERR) {
+		return true
+	}
+	if strings.HasPrefix(e, nats.ACCOUNT_AUTHENTICATION_EXPIRED_ERR) {
+		return true
+	}
+	return false
 }
 
-func (lu *LicenseUpdater) GetLicenseID() string {
-	lu.mu.Lock()
-	defer lu.mu.Unlock()
+func (c *NatsClient) GetLicenseID() string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.l.Status == v1alpha1.LicenseActive && time.Now().After(c.l.NotAfter.Time) {
+		license, _ := c.le.LoadLicense()
+		c.l = &license
+	}
+	return c.l.ID
+}
 
-	l := lu.License
-	if l.Status == v1alpha1.LicenseActive && time.Now().After(l.NotAfter.Time) {
-		license, _ := lu.le.LoadLicense()
-		lu.License = license
-		l = license
+func NewNatsConfig(cfg *rest.Config, clusterID string, LicenseFile string) *NatsClient {
+	return &NatsClient{
+		cfg:         cfg,
+		clusterID:   clusterID,
+		LicenseFile: LicenseFile,
 	}
-	return l.ID
 }
 
-func NewNatsConfig(cfg *rest.Config, clusterID string, LicenseFile string) (*NatsConfig, LicenseIDGetter, error) {
-	le, err := kubernetes.NewLicenseEnforcer(cfg, LicenseFile)
+func (c *NatsClient) connect() error {
+	le, err := kubernetes.NewLicenseEnforcer(c.cfg, c.LicenseFile)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 	license, licenseBytes := le.LoadLicense()
 	if license.Status != v1alpha1.LicenseActive {
-		return nil, nil, fmt.Errorf("license status is %s", license.Status)
+		return fmt.Errorf("license status is %s", license.Status)
 	}
 
 	opts := verifier.Options{
-		ClusterUID: clusterID,
+		ClusterUID: c.clusterID,
 		Features:   info.ProductName,
 		CACert:     []byte(info.LicenseCA),
 		License:    licenseBytes,
 	}
 	data, err := json.Marshal(opts)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
 	resp, err := http.Post(info.MustRegistrationAPIEndpoint(), "application/json", bytes.NewReader(data))
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 	defer resp.Body.Close()
 
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, nil, errors.New(resp.Status + ", " + string(body))
+		return errors.New(resp.Status + ", " + string(body))
 	}
 
 	var natscred NatsCredential
 	err = json.Unmarshal(body, &natscred)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
 	klog.V(5).InfoS("using event receiver", "address", natscred.Server, "subject", natscred.Subject, "licenseID", license.ID)
 
-	natscred.Client, err = NewConnection(license.ID, natscred)
+	nc, err := NewConnection(license.ID, natscred)
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 
-	return &natscred.NatsConfig, &LicenseUpdater{le: le, License: license}, nil
+	c.le = le
+	c.l = &license
+	c.nc = nc
+	c.Subject = natscred.Subject
+	c.Server = natscred.Server
+	return nil
 }
 
 // NewConnection creates a new NATS connection

lib/publisher.go

@@ -18,6 +18,7 @@ package lib
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	gosync "sync"
 	"time"
@@ -27,9 +28,7 @@ import (
 	cloudeventssdk "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding/format"
 	cloudevents "github.com/cloudevents/sdk-go/v2/event"
-	"github.com/nats-io/nats.go"
 	"go.bytebuilders.dev/license-verifier/info"
-	"gomodules.xyz/sync"
 	core "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -62,11 +61,7 @@ type Informer interface {
 type EventCreator func(obj client.Object) (*api.Event, error)
 
 type EventPublisher struct {
-	once    sync.Once
-	connect func() error
-
-	nats        *NatsConfig
-	lu          LicenseIDGetter
+	c           *NatsClient
 	mapper      discovery.ResourceMapper
 	createEvent EventCreator
 
@@ -75,47 +70,15 @@ type EventPublisher struct {
 }
 
 func NewEventPublisher(
-	nats *NatsConfig,
+	c *NatsClient,
 	mapper discovery.ResourceMapper,
 	fn EventCreator,
 ) *EventPublisher {
-	p := &EventPublisher{
+	return &EventPublisher{
+		c:           c,
 		mapper:      mapper,
 		createEvent: fn,
 	}
-	p.connect = func() error {
-		p.nats = nats
-		return nil
-	}
-	return p
-}
-
-func NewResilientEventPublisher(
-	fnConnect func() (*NatsConfig, LicenseIDGetter, error),
-	mapper discovery.ResourceMapper,
-	fnCreateEvent EventCreator,
-) *EventPublisher {
-	p := &EventPublisher{
-		mapper:      mapper,
-		createEvent: fnCreateEvent,
-	}
-	p.connect = func() error {
-		var err error
-		p.nats, p.lu, err = fnConnect()
-		if err != nil {
-			klog.V(5).InfoS("failed to connect with event receiver", "error", err)
-		}
-		return err
-	}
-	return p
-}
-
-func (p *EventPublisher) NatsClient() (*nats.Conn, error) {
-	p.once.Do(p.connect)
-	if p.nats == nil {
-		return nil, fmt.Errorf("not connected to nats")
-	}
-	return p.nats.Client, nil
 }
 
 func (p *EventPublisher) Publish(ev *api.Event, et api.EventType) error {
@@ -145,7 +108,7 @@ func (p *EventPublisher) Publish(ev *api.Event, et api.EventType) error {
 	defer cancel()
 
 	for {
-		_, err = p.nats.Client.Request(p.nats.Subject, data, natsRequestTimeout)
+		_, err = p.c.Request(data, natsRequestTimeout)
 		if err == nil {
 			cancel()
 		} else {
@@ -154,10 +117,10 @@ func (p *EventPublisher) Publish(ev *api.Event, et api.EventType) error {
 
 		select {
 		case <-ctx.Done():
-			if ctx.Err() == context.DeadlineExceeded {
+			if errors.Is(ctx.Err(), context.DeadlineExceeded) {
 				klog.V(5).Infof("failed to send event : %s", string(data))
-			} else if ctx.Err() == context.Canceled {
-				klog.V(5).Infof("Published event `%s` to channel `%s` and acknowledged", et, p.nats.Subject)
+			} else if errors.Is(ctx.Err(), context.Canceled) {
+				klog.V(5).Infof("Published event `%s` to channel `%s` and acknowledged", et, p.c.Subject)
 			}
 			return nil
 		default:
@@ -183,13 +146,7 @@ func (p *EventPublisher) ForGVK(informer Informer, gvk schema.GroupVersionKind)
 			if err != nil {
 				return nil, err
 			}
-
-			p.once.Do(p.connect)
-			if p.nats == nil {
-				return nil, fmt.Errorf("not connected to nats")
-			}
-			ev.LicenseID = p.lu.GetLicenseID()
-
+			ev.LicenseID = p.c.GetLicenseID()
 			return ev, nil
 		},
 	}
@@ -255,12 +212,7 @@ func (p *EventPublisher) setupSiteInfoPublisher(cfg *rest.Config, kc kubernetes.
 		identitylib.RefreshNodeStats(p.si, nodes)
 		p.siMutex.Unlock()
 
-		p.once.Do(p.connect)
-		if p.nats == nil {
-			return nil, fmt.Errorf("not connected to nats")
-		}
-
-		licenseID := p.lu.GetLicenseID()
+		licenseID := p.c.GetLicenseID()
 		p.si.Product.LicenseID = licenseID
 		p.si.Name = fmt.Sprintf("%s.%s", licenseID, p.si.Product.ProductName)
 		ev := &api.Event{

vendor/go.bytebuilders.dev/license-verifier/apis/licenses/v1alpha1/helper.go

@@ -17,7 +17,11 @@ limitations under the License.
 package v1alpha1
 
 func (l License) DisableAnalytics() bool {
-	return len(l.FeatureFlags) > 0 && l.FeatureFlags["DisableAnalytics"] == "true"
+	return len(l.FeatureFlags) > 0 && l.FeatureFlags[FeatureDisableAnalytics] == "true"
+}
+
+func (l License) EnableClientBilling() bool {
+	return len(l.FeatureFlags) > 0 && l.FeatureFlags[FeatureEnableClientBilling] == "true"
 }
 
 func (i *License) Less(j *License) bool {