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

	discoverysnapshotInView = &view.View{
		Name:        "discovery.supergloo.solo.io_snap_emitter/snap_in",
		Measure:     mDiscoverySnapshotIn,
		Description: "The number of snapshots updates coming in",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
	discoverysnapshotOutView = &view.View{
		Name:        "discovery.supergloo.solo.io/snap_emitter/snap_out",
		Measure:     mDiscoverySnapshotOut,
		Description: "The number of snapshots updates going out",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
)

func init() {
	view.Register(discoverysnapshotInView, discoverysnapshotOutView)
}

type DiscoveryEmitter interface {
	Register() error
	Pod() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
	Upstream() gloo_solo_io.UpstreamClient
	TlsSecret() TlsSecretClient
	Deployment() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentClient
	Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *DiscoverySnapshot, <-chan error, error)
}

func NewDiscoveryEmitter(podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, upstreamClient gloo_solo_io.UpstreamClient, tlsSecretClient TlsSecretClient, deploymentClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentClient) DiscoveryEmitter {
	return NewDiscoveryEmitterWithEmit(podClient, upstreamClient, tlsSecretClient, deploymentClient, make(chan struct{}))
}

func NewDiscoveryEmitterWithEmit(podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, upstreamClient gloo_solo_io.UpstreamClient, tlsSecretClient TlsSecretClient, deploymentClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentClient, emit <-chan struct{}) DiscoveryEmitter {
	return &discoveryEmitter{
		pod:        podClient,
		upstream:   upstreamClient,
		tlsSecret:  tlsSecretClient,
		deployment: deploymentClient,
		forceEmit:  emit,
	}
}

type discoveryEmitter struct {
	forceEmit  <-chan struct{}
	pod        github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
	upstream   gloo_solo_io.UpstreamClient
	tlsSecret  TlsSecretClient
	deployment github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentClient
}

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

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

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

func (c *discoveryEmitter) TlsSecret() TlsSecretClient {
	return c.tlsSecret
}

func (c *discoveryEmitter) Deployment() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentClient {
	return c.deployment
}

func (c *discoveryEmitter) Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *DiscoverySnapshot, <-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 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)
	/* Create channel for TlsSecret */
	type tlsSecretListWithNamespace struct {
		list      TlsSecretList
		namespace string
	}
	tlsSecretChan := make(chan tlsSecretListWithNamespace)
	/* Create channel for Deployment */
	type deploymentListWithNamespace struct {
		list      github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentList
		namespace string
	}
	deploymentChan := make(chan deploymentListWithNamespace)

	for _, namespace := range watchNamespaces {
		/* 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)
		/* Setup namespaced watch for TlsSecret */
		tlsSecretNamespacesChan, tlsSecretErrs, err := c.tlsSecret.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting TlsSecret watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, tlsSecretErrs, namespace+"-tlssecrets")
		}(namespace)
		/* Setup namespaced watch for Deployment */
		deploymentNamespacesChan, deploymentErrs, err := c.deployment.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting Deployment watch")
		}

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

		/* Watch for changes and update snapshot */
		go func(namespace string) {
			for {
				select {
				case <-ctx.Done():
					return
				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}:
					}
				case tlsSecretList := <-tlsSecretNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case tlsSecretChan <- tlsSecretListWithNamespace{list: tlsSecretList, namespace: namespace}:
					}
				case deploymentList := <-deploymentNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case deploymentChan <- deploymentListWithNamespace{list: deploymentList, namespace: namespace}:
					}
				}
			}
		}(namespace)
	}

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

			stats.Record(ctx, mDiscoverySnapshotOut.M(1))
			originalSnapshot = currentSnapshot.Clone()
			sentSnapshot := currentSnapshot.Clone()
			snapshots <- &sentSnapshot
		}
		podsByNamespace := make(map[string]github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodList)
		upstreamsByNamespace := make(map[string]gloo_solo_io.UpstreamList)
		tlssecretsByNamespace := make(map[string]TlsSecretList)
		deploymentsByNamespace := make(map[string]github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentList)

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

				namespace := podNamespacedList.namespace

				// merge lists by namespace
				podsByNamespace[namespace] = podNamespacedList.list
				var podList github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodList
				for _, pods := range podsByNamespace {
					podList = append(podList, pods...)
				}
				currentSnapshot.Pods = podList.Sort()
			case upstreamNamespacedList := <-upstreamChan:
				record()

				namespace := upstreamNamespacedList.namespace

				// merge lists by namespace
				upstreamsByNamespace[namespace] = upstreamNamespacedList.list
				var upstreamList gloo_solo_io.UpstreamList
				for _, upstreams := range upstreamsByNamespace {
					upstreamList = append(upstreamList, upstreams...)
				}
				currentSnapshot.Upstreams = upstreamList.Sort()
			case tlsSecretNamespacedList := <-tlsSecretChan:
				record()

				namespace := tlsSecretNamespacedList.namespace

				// merge lists by namespace
				tlssecretsByNamespace[namespace] = tlsSecretNamespacedList.list
				var tlsSecretList TlsSecretList
				for _, tlssecrets := range tlssecretsByNamespace {
					tlsSecretList = append(tlsSecretList, tlssecrets...)
				}
				currentSnapshot.Tlssecrets = tlsSecretList.Sort()
			case deploymentNamespacedList := <-deploymentChan:
				record()

				namespace := deploymentNamespacedList.namespace

				// merge lists by namespace
				deploymentsByNamespace[namespace] = deploymentNamespacedList.list
				var deploymentList github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.DeploymentList
				for _, deployments := range deploymentsByNamespace {
					deploymentList = append(deploymentList, deployments...)
				}
				currentSnapshot.Deployments = deploymentList.Sort()
			}
		}
	}()
	return snapshots, errs, nil
}