appmesh_discovery_snapshot_emitter.sk.go

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

package v1

import (
	"sync"
	"time"

	gloo_solo_io "github.com/solo-io/gloo/projects/gloo/pkg/api/v1"
	github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes "github.com/solo-io/solo-kit/pkg/api/v1/resources/common/kubernetes"

	"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 (
	mAppmeshDiscoverySnapshotIn  = stats.Int64("appmeshDiscovery.supergloo.solo.io/snap_emitter/snap_in", "The number of snapshots in", "1")
	mAppmeshDiscoverySnapshotOut = stats.Int64("appmeshDiscovery.supergloo.solo.io/snap_emitter/snap_out", "The number of snapshots out", "1")

	appmeshDiscoverysnapshotInView = &view.View{
		Name:        "appmeshDiscovery.supergloo.solo.io_snap_emitter/snap_in",
		Measure:     mAppmeshDiscoverySnapshotIn,
		Description: "The number of snapshots updates coming in",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
	appmeshDiscoverysnapshotOutView = &view.View{
		Name:        "appmeshDiscovery.supergloo.solo.io/snap_emitter/snap_out",
		Measure:     mAppmeshDiscoverySnapshotOut,
		Description: "The number of snapshots updates going out",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
)

func init() {
	view.Register(appmeshDiscoverysnapshotInView, appmeshDiscoverysnapshotOutView)
}

type AppmeshDiscoveryEmitter interface {
	Register() error
	Mesh() MeshClient
	Pod() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
	Upstream() gloo_solo_io.UpstreamClient
	Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *AppmeshDiscoverySnapshot, <-chan error, error)
}

func NewAppmeshDiscoveryEmitter(meshClient MeshClient, podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, upstreamClient gloo_solo_io.UpstreamClient) AppmeshDiscoveryEmitter {
	return NewAppmeshDiscoveryEmitterWithEmit(meshClient, podClient, upstreamClient, make(chan struct{}))
}

func NewAppmeshDiscoveryEmitterWithEmit(meshClient MeshClient, podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, upstreamClient gloo_solo_io.UpstreamClient, emit <-chan struct{}) AppmeshDiscoveryEmitter {
	return &appmeshDiscoveryEmitter{
		mesh:      meshClient,
		pod:       podClient,
		upstream:  upstreamClient,
		forceEmit: emit,
	}
}

type appmeshDiscoveryEmitter struct {
	forceEmit <-chan struct{}
	mesh      MeshClient
	pod       github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
	upstream  gloo_solo_io.UpstreamClient
}

func (c *appmeshDiscoveryEmitter) Register() error {
	if err := c.mesh.Register(); err != nil {
		return err
	}
	if err := c.pod.Register(); err != nil {
		return err
	}
	if err := c.upstream.Register(); err != nil {
		return err
	}
	return nil
}

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

func (c *appmeshDiscoveryEmitter) Pod() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient {
	return c.pod
}

func (c *appmeshDiscoveryEmitter) Upstream() gloo_solo_io.UpstreamClient {
	return c.upstream
}

func (c *appmeshDiscoveryEmitter) Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *AppmeshDiscoverySnapshot, <-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 Pod */
	type podListWithNamespace struct {
		list      github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodList
		namespace string
	}
	podChan := make(chan podListWithNamespace)
	/* Create channel for Upstream */
	type upstreamListWithNamespace struct {
		list      gloo_solo_io.UpstreamList
		namespace string
	}
	upstreamChan := make(chan upstreamListWithNamespace)

	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 Pod */
		podNamespacesChan, podErrs, err := c.pod.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting Pod watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, podErrs, namespace+"-pods")
		}(namespace)
		/* Setup namespaced watch for Upstream */
		upstreamNamespacesChan, upstreamErrs, err := c.upstream.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting Upstream watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, upstreamErrs, namespace+"-upstreams")
		}(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 podList := <-podNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case podChan <- podListWithNamespace{list: podList, namespace: namespace}:
					}
				case upstreamList := <-upstreamNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case upstreamChan <- upstreamListWithNamespace{list: upstreamList, namespace: namespace}:
					}
				}
			}
		}(namespace)
	}

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

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

		for {
			record := func() { stats.Record(ctx, mAppmeshDiscoverySnapshotIn.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 podNamespacedList := <-podChan:
				record()

				namespace := podNamespacedList.namespace
				podList := podNamespacedList.list

				currentSnapshot.Pods[namespace] = podList
			case upstreamNamespacedList := <-upstreamChan:
				record()

				namespace := upstreamNamespacedList.namespace
				upstreamList := upstreamNamespacedList.list

				currentSnapshot.Upstreams[namespace] = upstreamList
			}
		}
	}()
	return snapshots, errs, nil
}