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

	configsnapshotInView = &view.View{
		Name:        "config.supergloo.solo.io_snap_emitter/snap_in",
		Measure:     mConfigSnapshotIn,
		Description: "The number of snapshots updates coming in",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
	configsnapshotOutView = &view.View{
		Name:        "config.supergloo.solo.io/snap_emitter/snap_out",
		Measure:     mConfigSnapshotOut,
		Description: "The number of snapshots updates going out",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
)

func init() {
	view.Register(configsnapshotInView, configsnapshotOutView)
}

type ConfigEmitter interface {
	Register() error
	Mesh() MeshClient
	MeshIngress() MeshIngressClient
	MeshGroup() MeshGroupClient
	RoutingRule() RoutingRuleClient
	SecurityRule() SecurityRuleClient
	TlsSecret() TlsSecretClient
	Upstream() gloo_solo_io.UpstreamClient
	Pod() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
	Service() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceClient
	CustomResourceDefinition() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionClient
	Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *ConfigSnapshot, <-chan error, error)
}

func NewConfigEmitter(meshClient MeshClient, meshIngressClient MeshIngressClient, meshGroupClient MeshGroupClient, routingRuleClient RoutingRuleClient, securityRuleClient SecurityRuleClient, tlsSecretClient TlsSecretClient, upstreamClient gloo_solo_io.UpstreamClient, podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, serviceClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceClient, customResourceDefinitionClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionClient) ConfigEmitter {
	return NewConfigEmitterWithEmit(meshClient, meshIngressClient, meshGroupClient, routingRuleClient, securityRuleClient, tlsSecretClient, upstreamClient, podClient, serviceClient, customResourceDefinitionClient, make(chan struct{}))
}

func NewConfigEmitterWithEmit(meshClient MeshClient, meshIngressClient MeshIngressClient, meshGroupClient MeshGroupClient, routingRuleClient RoutingRuleClient, securityRuleClient SecurityRuleClient, tlsSecretClient TlsSecretClient, upstreamClient gloo_solo_io.UpstreamClient, podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, serviceClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceClient, customResourceDefinitionClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionClient, emit <-chan struct{}) ConfigEmitter {
	return &configEmitter{
		mesh:                     meshClient,
		meshIngress:              meshIngressClient,
		meshGroup:                meshGroupClient,
		routingRule:              routingRuleClient,
		securityRule:             securityRuleClient,
		tlsSecret:                tlsSecretClient,
		upstream:                 upstreamClient,
		pod:                      podClient,
		service:                  serviceClient,
		customResourceDefinition: customResourceDefinitionClient,
		forceEmit:                emit,
	}
}

type configEmitter struct {
	forceEmit                <-chan struct{}
	mesh                     MeshClient
	meshIngress              MeshIngressClient
	meshGroup                MeshGroupClient
	routingRule              RoutingRuleClient
	securityRule             SecurityRuleClient
	tlsSecret                TlsSecretClient
	upstream                 gloo_solo_io.UpstreamClient
	pod                      github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
	service                  github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceClient
	customResourceDefinition github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionClient
}

func (c *configEmitter) Register() error {
	if err := c.mesh.Register(); err != nil {
		return err
	}
	if err := c.meshIngress.Register(); err != nil {
		return err
	}
	if err := c.meshGroup.Register(); err != nil {
		return err
	}
	if err := c.routingRule.Register(); err != nil {
		return err
	}
	if err := c.securityRule.Register(); err != nil {
		return err
	}
	if err := c.tlsSecret.Register(); err != nil {
		return err
	}
	if err := c.upstream.Register(); err != nil {
		return err
	}
	if err := c.pod.Register(); err != nil {
		return err
	}
	if err := c.service.Register(); err != nil {
		return err
	}
	if err := c.customResourceDefinition.Register(); err != nil {
		return err
	}
	return nil
}

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

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

func (c *configEmitter) MeshGroup() MeshGroupClient {
	return c.meshGroup
}

func (c *configEmitter) RoutingRule() RoutingRuleClient {
	return c.routingRule
}

func (c *configEmitter) SecurityRule() SecurityRuleClient {
	return c.securityRule
}

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

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

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

func (c *configEmitter) Service() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceClient {
	return c.service
}

func (c *configEmitter) CustomResourceDefinition() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionClient {
	return c.customResourceDefinition
}

func (c *configEmitter) Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *ConfigSnapshot, <-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)
	/* Create channel for MeshGroup */
	type meshGroupListWithNamespace struct {
		list      MeshGroupList
		namespace string
	}
	meshGroupChan := make(chan meshGroupListWithNamespace)
	/* Create channel for RoutingRule */
	type routingRuleListWithNamespace struct {
		list      RoutingRuleList
		namespace string
	}
	routingRuleChan := make(chan routingRuleListWithNamespace)
	/* Create channel for SecurityRule */
	type securityRuleListWithNamespace struct {
		list      SecurityRuleList
		namespace string
	}
	securityRuleChan := make(chan securityRuleListWithNamespace)
	/* Create channel for TlsSecret */
	type tlsSecretListWithNamespace struct {
		list      TlsSecretList
		namespace string
	}
	tlsSecretChan := make(chan tlsSecretListWithNamespace)
	/* Create channel for Upstream */
	type upstreamListWithNamespace struct {
		list      gloo_solo_io.UpstreamList
		namespace string
	}
	upstreamChan := make(chan upstreamListWithNamespace)
	/* 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 Service */
	type serviceListWithNamespace struct {
		list      github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceList
		namespace string
	}
	serviceChan := make(chan serviceListWithNamespace)
	/* Create channel for CustomResourceDefinition */
	type customResourceDefinitionListWithNamespace struct {
		list      github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionList
		namespace string
	}
	customResourceDefinitionChan := make(chan customResourceDefinitionListWithNamespace)

	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)
		/* Setup namespaced watch for MeshGroup */
		meshGroupNamespacesChan, meshGroupErrs, err := c.meshGroup.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting MeshGroup watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, meshGroupErrs, namespace+"-meshgroups")
		}(namespace)
		/* Setup namespaced watch for RoutingRule */
		routingRuleNamespacesChan, routingRuleErrs, err := c.routingRule.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting RoutingRule watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, routingRuleErrs, namespace+"-routingrules")
		}(namespace)
		/* Setup namespaced watch for SecurityRule */
		securityRuleNamespacesChan, securityRuleErrs, err := c.securityRule.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting SecurityRule watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, securityRuleErrs, namespace+"-securityrules")
		}(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 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 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 Service */
		serviceNamespacesChan, serviceErrs, err := c.service.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting Service watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, serviceErrs, namespace+"-services")
		}(namespace)
		/* Setup namespaced watch for CustomResourceDefinition */
		customResourceDefinitionNamespacesChan, customResourceDefinitionErrs, err := c.customResourceDefinition.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting CustomResourceDefinition watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, customResourceDefinitionErrs, namespace+"-customresourcedefinition")
		}(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}:
					}
				case meshGroupList := <-meshGroupNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case meshGroupChan <- meshGroupListWithNamespace{list: meshGroupList, namespace: namespace}:
					}
				case routingRuleList := <-routingRuleNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case routingRuleChan <- routingRuleListWithNamespace{list: routingRuleList, namespace: namespace}:
					}
				case securityRuleList := <-securityRuleNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case securityRuleChan <- securityRuleListWithNamespace{list: securityRuleList, namespace: namespace}:
					}
				case tlsSecretList := <-tlsSecretNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case tlsSecretChan <- tlsSecretListWithNamespace{list: tlsSecretList, namespace: namespace}:
					}
				case upstreamList := <-upstreamNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case upstreamChan <- upstreamListWithNamespace{list: upstreamList, namespace: namespace}:
					}
				case podList := <-podNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case podChan <- podListWithNamespace{list: podList, namespace: namespace}:
					}
				case serviceList := <-serviceNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case serviceChan <- serviceListWithNamespace{list: serviceList, namespace: namespace}:
					}
				case customResourceDefinitionList := <-customResourceDefinitionNamespacesChan:
					select {
					case <-ctx.Done():
						return
					case customResourceDefinitionChan <- customResourceDefinitionListWithNamespace{list: customResourceDefinitionList, namespace: namespace}:
					}
				}
			}
		}(namespace)
	}

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

			stats.Record(ctx, mConfigSnapshotOut.M(1))
			originalSnapshot = currentSnapshot.Clone()
			sentSnapshot := currentSnapshot.Clone()
			snapshots <- &sentSnapshot
		}
		meshesByNamespace := make(map[string]MeshList)
		meshingressesByNamespace := make(map[string]MeshIngressList)
		meshgroupsByNamespace := make(map[string]MeshGroupList)
		routingrulesByNamespace := make(map[string]RoutingRuleList)
		securityrulesByNamespace := make(map[string]SecurityRuleList)
		tlssecretsByNamespace := make(map[string]TlsSecretList)
		upstreamsByNamespace := make(map[string]gloo_solo_io.UpstreamList)
		podsByNamespace := make(map[string]github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodList)
		servicesByNamespace := make(map[string]github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceList)
		customresourcedefinitionByNamespace := make(map[string]github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionList)

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

				// merge lists by namespace
				meshesByNamespace[namespace] = meshNamespacedList.list
				var meshList MeshList
				for _, meshes := range meshesByNamespace {
					meshList = append(meshList, meshes...)
				}
				currentSnapshot.Meshes = meshList.Sort()
			case meshIngressNamespacedList := <-meshIngressChan:
				record()

				namespace := meshIngressNamespacedList.namespace

				// merge lists by namespace
				meshingressesByNamespace[namespace] = meshIngressNamespacedList.list
				var meshIngressList MeshIngressList
				for _, meshingresses := range meshingressesByNamespace {
					meshIngressList = append(meshIngressList, meshingresses...)
				}
				currentSnapshot.Meshingresses = meshIngressList.Sort()
			case meshGroupNamespacedList := <-meshGroupChan:
				record()

				namespace := meshGroupNamespacedList.namespace

				// merge lists by namespace
				meshgroupsByNamespace[namespace] = meshGroupNamespacedList.list
				var meshGroupList MeshGroupList
				for _, meshgroups := range meshgroupsByNamespace {
					meshGroupList = append(meshGroupList, meshgroups...)
				}
				currentSnapshot.Meshgroups = meshGroupList.Sort()
			case routingRuleNamespacedList := <-routingRuleChan:
				record()

				namespace := routingRuleNamespacedList.namespace

				// merge lists by namespace
				routingrulesByNamespace[namespace] = routingRuleNamespacedList.list
				var routingRuleList RoutingRuleList
				for _, routingrules := range routingrulesByNamespace {
					routingRuleList = append(routingRuleList, routingrules...)
				}
				currentSnapshot.Routingrules = routingRuleList.Sort()
			case securityRuleNamespacedList := <-securityRuleChan:
				record()

				namespace := securityRuleNamespacedList.namespace

				// merge lists by namespace
				securityrulesByNamespace[namespace] = securityRuleNamespacedList.list
				var securityRuleList SecurityRuleList
				for _, securityrules := range securityrulesByNamespace {
					securityRuleList = append(securityRuleList, securityrules...)
				}
				currentSnapshot.Securityrules = securityRuleList.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 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 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 serviceNamespacedList := <-serviceChan:
				record()

				namespace := serviceNamespacedList.namespace

				// merge lists by namespace
				servicesByNamespace[namespace] = serviceNamespacedList.list
				var serviceList github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.ServiceList
				for _, services := range servicesByNamespace {
					serviceList = append(serviceList, services...)
				}
				currentSnapshot.Services = serviceList.Sort()
			case customResourceDefinitionNamespacedList := <-customResourceDefinitionChan:
				record()

				namespace := customResourceDefinitionNamespacedList.namespace

				// merge lists by namespace
				customresourcedefinitionByNamespace[namespace] = customResourceDefinitionNamespacedList.list
				var customResourceDefinitionList github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.CustomResourceDefinitionList
				for _, customresourcedefinition := range customresourcedefinitionByNamespace {
					customResourceDefinitionList = append(customResourceDefinitionList, customresourcedefinition...)
				}
				currentSnapshot.Customresourcedefinition = customResourceDefinitionList.Sort()
			}
		}
	}()
	return snapshots, errs, nil
}