registration_snapshot_emitter.sk.go

// Code generated by solo-kit. DO NOT EDIT.

package v1

import (
	"sync"
	"time"

	"go.opencensus.io/stats"
	"go.opencensus.io/stats/view"
	"go.opencensus.io/tag"

	"github.com/solo-io/go-utils/errutils"
	"github.com/solo-io/solo-kit/pkg/api/v1/clients"
	"github.com/solo-io/solo-kit/pkg/errors"
)

var (
	mRegistrationSnapshotIn  = stats.Int64("registration.supergloo.solo.io/snap_emitter/snap_in", "The number of snapshots in", "1")
	mRegistrationSnapshotOut = stats.Int64("registration.supergloo.solo.io/snap_emitter/snap_out", "The number of snapshots out", "1")

	registrationsnapshotInView = &view.View{
		Name:        "registration.supergloo.solo.io_snap_emitter/snap_in",
		Measure:     mRegistrationSnapshotIn,
		Description: "The number of snapshots updates coming in",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
	registrationsnapshotOutView = &view.View{
		Name:        "registration.supergloo.solo.io/snap_emitter/snap_out",
		Measure:     mRegistrationSnapshotOut,
		Description: "The number of snapshots updates going out",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
)

func init() {
	view.Register(registrationsnapshotInView, registrationsnapshotOutView)
}

type RegistrationEmitter interface {
	Register() error
	Mesh() MeshClient
	MeshIngress() MeshIngressClient
	Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *RegistrationSnapshot, <-chan error, error)
}

func NewRegistrationEmitter(meshClient MeshClient, meshIngressClient MeshIngressClient) RegistrationEmitter {
	return NewRegistrationEmitterWithEmit(meshClient, meshIngressClient, make(chan struct{}))
}

func NewRegistrationEmitterWithEmit(meshClient MeshClient, meshIngressClient MeshIngressClient, emit <-chan struct{}) RegistrationEmitter {
	return &registrationEmitter{
		mesh:        meshClient,
		meshIngress: meshIngressClient,
		forceEmit:   emit,
	}
}

type registrationEmitter struct {
	forceEmit   <-chan struct{}
	mesh        MeshClient
	meshIngress MeshIngressClient
}

func (c *registrationEmitter) Register() error {
	if err := c.mesh.Register(); err != nil {
		return err
	}
	if err := c.meshIngress.Register(); err != nil {
		return err
	}
	return nil
}

func (c *registrationEmitter) Mesh() MeshClient {
	return c.mesh
}

func (c *registrationEmitter) MeshIngress() MeshIngressClient {
	return c.meshIngress
}

func (c *registrationEmitter) Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *RegistrationSnapshot, <-chan error, error) {

	if len(watchNamespaces) == 0 {
		watchNamespaces = []string{""}
	}

	for _, ns := range watchNamespaces {
		if ns == "" && len(watchNamespaces) > 1 {
			return nil, nil, errors.Errorf("the \"\" namespace is used to watch all namespaces. Snapshots can either be tracked for " +
				"specific namespaces or \"\" AllNamespaces, but not both.")
		}
	}

	errs := make(chan error)
	var done sync.WaitGroup
	ctx := opts.Ctx
	/* Create channel for Mesh */
	type meshListWithNamespace struct {
		list      MeshList
		namespace string
	}
	meshChan := make(chan meshListWithNamespace)
	/* Create channel for MeshIngress */
	type meshIngressListWithNamespace struct {
		list      MeshIngressList
		namespace string
	}
	meshIngressChan := make(chan meshIngressListWithNamespace)

	for _, namespace := range watchNamespaces {
		/* Setup namespaced watch for Mesh */
		meshNamespacesChan, meshErrs, err := c.mesh.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting Mesh watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, meshErrs, namespace+"-meshes")
		}(namespace)
		/* Setup namespaced watch for MeshIngress */
		meshIngressNamespacesChan, meshIngressErrs, err := c.meshIngress.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting MeshIngress watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, meshIngressErrs, namespace+"-meshingresses")
		}(namespace)

		/* Watch for changes and update snapshot */
		go func(namespace string) {
			for {
				select {
				case <-ctx.Done():
					return
				case meshList := <-meshNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case meshChan <- meshListWithNamespace{list: meshList, namespace: namespace}:
					}
				case meshIngressList := <-meshIngressNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case meshIngressChan <- meshIngressListWithNamespace{list: meshIngressList, namespace: namespace}:
					}
				}
			}
		}(namespace)
	}

	snapshots := make(chan *RegistrationSnapshot)
	go func() {
		originalSnapshot := RegistrationSnapshot{}
		currentSnapshot := originalSnapshot.Clone()
		timer := time.NewTicker(time.Second * 1)
		sync := func() {
			if originalSnapshot.Hash() == currentSnapshot.Hash() {
				return
			}

			stats.Record(ctx, mRegistrationSnapshotOut.M(1))
			originalSnapshot = currentSnapshot.Clone()
			sentSnapshot := currentSnapshot.Clone()
			snapshots <- &sentSnapshot
		}

		for {
			record := func() { stats.Record(ctx, mRegistrationSnapshotIn.M(1)) }

			select {
			case <-timer.C:
				sync()
			case <-ctx.Done():
				close(snapshots)
				done.Wait()
				close(errs)
				return
			case <-c.forceEmit:
				sentSnapshot := currentSnapshot.Clone()
				snapshots <- &sentSnapshot
			case meshNamespacedList := <-meshChan:
				record()

				namespace := meshNamespacedList.namespace
				meshList := meshNamespacedList.list

				currentSnapshot.Meshes[namespace] = meshList
			case meshIngressNamespacedList := <-meshIngressChan:
				record()

				namespace := meshIngressNamespacedList.namespace
				meshIngressList := meshIngressNamespacedList.list

				currentSnapshot.Meshingresses[namespace] = meshIngressList
			}
		}
	}()
	return snapshots, errs, nil
}