testing_snapshot_emitter.sk.go

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

package v1

import (
	"sync"
	"time"

	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"
	"go.uber.org/zap"

	"github.com/solo-io/solo-kit/pkg/api/v1/clients"
	"github.com/solo-io/solo-kit/pkg/errors"
	skstats "github.com/solo-io/solo-kit/pkg/stats"

	"github.com/solo-io/go-utils/contextutils"
	"github.com/solo-io/go-utils/errutils"
)

var (
	// Deprecated. See mTestingResourcesIn
	mTestingSnapshotIn = stats.Int64("testing.solo.io/emitter/snap_in", "Deprecated. Use testing.solo.io/emitter/resources_in. The number of snapshots in", "1")

	// metrics for emitter
	mTestingResourcesIn    = stats.Int64("testing.solo.io/emitter/resources_in", "The number of resource lists received on open watch channels", "1")
	mTestingSnapshotOut    = stats.Int64("testing.solo.io/emitter/snap_out", "The number of snapshots out", "1")
	mTestingSnapshotMissed = stats.Int64("testing.solo.io/emitter/snap_missed", "The number of snapshots missed", "1")

	// views for emitter
	// deprecated: see testingResourcesInView
	testingsnapshotInView = &view.View{
		Name:        "testing.solo.io/emitter/snap_in",
		Measure:     mTestingSnapshotIn,
		Description: "Deprecated. Use testing.solo.io/emitter/resources_in. The number of snapshots updates coming in.",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}

	testingResourcesInView = &view.View{
		Name:        "testing.solo.io/emitter/resources_in",
		Measure:     mTestingResourcesIn,
		Description: "The number of resource lists received on open watch channels",
		Aggregation: view.Count(),
		TagKeys: []tag.Key{
			skstats.NamespaceKey,
			skstats.ResourceKey,
		},
	}
	testingsnapshotOutView = &view.View{
		Name:        "testing.solo.io/emitter/snap_out",
		Measure:     mTestingSnapshotOut,
		Description: "The number of snapshots updates going out",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
	testingsnapshotMissedView = &view.View{
		Name:        "testing.solo.io/emitter/snap_missed",
		Measure:     mTestingSnapshotMissed,
		Description: "The number of snapshots updates going missed. this can happen in heavy load. missed snapshot will be re-tried after a second.",
		Aggregation: view.Count(),
		TagKeys:     []tag.Key{},
	}
)

func init() {
	view.Register(
		testingsnapshotInView,
		testingsnapshotOutView,
		testingsnapshotMissedView,
		testingResourcesInView,
	)
}

type TestingSnapshotEmitter interface {
	Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *TestingSnapshot, <-chan error, error)
}

type TestingEmitter interface {
	TestingSnapshotEmitter
	Register() error
	SimpleMockResource() SimpleMockResourceClient
	MockResource() MockResourceClient
	FakeResource() FakeResourceClient
	AnotherMockResource() AnotherMockResourceClient
	ClusterResource() ClusterResourceClient
	MockCustomType() MockCustomTypeClient
	MockCustomSpecHashType() MockCustomSpecHashTypeClient
	Pod() github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
}

func NewTestingEmitter(simpleMockResourceClient SimpleMockResourceClient, mockResourceClient MockResourceClient, fakeResourceClient FakeResourceClient, anotherMockResourceClient AnotherMockResourceClient, clusterResourceClient ClusterResourceClient, mockCustomTypeClient MockCustomTypeClient, mockCustomSpecHashTypeClient MockCustomSpecHashTypeClient, podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient) TestingEmitter {
	return NewTestingEmitterWithEmit(simpleMockResourceClient, mockResourceClient, fakeResourceClient, anotherMockResourceClient, clusterResourceClient, mockCustomTypeClient, mockCustomSpecHashTypeClient, podClient, make(chan struct{}))
}

func NewTestingEmitterWithEmit(simpleMockResourceClient SimpleMockResourceClient, mockResourceClient MockResourceClient, fakeResourceClient FakeResourceClient, anotherMockResourceClient AnotherMockResourceClient, clusterResourceClient ClusterResourceClient, mockCustomTypeClient MockCustomTypeClient, mockCustomSpecHashTypeClient MockCustomSpecHashTypeClient, podClient github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient, emit <-chan struct{}) TestingEmitter {
	return &testingEmitter{
		simpleMockResource:     simpleMockResourceClient,
		mockResource:           mockResourceClient,
		fakeResource:           fakeResourceClient,
		anotherMockResource:    anotherMockResourceClient,
		clusterResource:        clusterResourceClient,
		mockCustomType:         mockCustomTypeClient,
		mockCustomSpecHashType: mockCustomSpecHashTypeClient,
		pod:                    podClient,
		forceEmit:              emit,
	}
}

type testingEmitter struct {
	forceEmit              <-chan struct{}
	simpleMockResource     SimpleMockResourceClient
	mockResource           MockResourceClient
	fakeResource           FakeResourceClient
	anotherMockResource    AnotherMockResourceClient
	clusterResource        ClusterResourceClient
	mockCustomType         MockCustomTypeClient
	mockCustomSpecHashType MockCustomSpecHashTypeClient
	pod                    github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodClient
}

func (c *testingEmitter) Register() error {
	if err := c.simpleMockResource.Register(); err != nil {
		return err
	}
	if err := c.mockResource.Register(); err != nil {
		return err
	}
	if err := c.fakeResource.Register(); err != nil {
		return err
	}
	if err := c.anotherMockResource.Register(); err != nil {
		return err
	}
	if err := c.clusterResource.Register(); err != nil {
		return err
	}
	if err := c.mockCustomType.Register(); err != nil {
		return err
	}
	if err := c.mockCustomSpecHashType.Register(); err != nil {
		return err
	}
	if err := c.pod.Register(); err != nil {
		return err
	}
	return nil
}

func (c *testingEmitter) SimpleMockResource() SimpleMockResourceClient {
	return c.simpleMockResource
}

func (c *testingEmitter) MockResource() MockResourceClient {
	return c.mockResource
}

func (c *testingEmitter) FakeResource() FakeResourceClient {
	return c.fakeResource
}

func (c *testingEmitter) AnotherMockResource() AnotherMockResourceClient {
	return c.anotherMockResource
}

func (c *testingEmitter) ClusterResource() ClusterResourceClient {
	return c.clusterResource
}

func (c *testingEmitter) MockCustomType() MockCustomTypeClient {
	return c.mockCustomType
}

func (c *testingEmitter) MockCustomSpecHashType() MockCustomSpecHashTypeClient {
	return c.mockCustomSpecHashType
}

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

func (c *testingEmitter) Snapshots(watchNamespaces []string, opts clients.WatchOpts) (<-chan *TestingSnapshot, <-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 SimpleMockResource */
	type simpleMockResourceListWithNamespace struct {
		list      SimpleMockResourceList
		namespace string
	}
	simpleMockResourceChan := make(chan simpleMockResourceListWithNamespace)

	var initialSimpleMockResourceList SimpleMockResourceList
	/* Create channel for MockResource */
	type mockResourceListWithNamespace struct {
		list      MockResourceList
		namespace string
	}
	mockResourceChan := make(chan mockResourceListWithNamespace)

	var initialMockResourceList MockResourceList
	/* Create channel for FakeResource */
	type fakeResourceListWithNamespace struct {
		list      FakeResourceList
		namespace string
	}
	fakeResourceChan := make(chan fakeResourceListWithNamespace)

	var initialFakeResourceList FakeResourceList
	/* Create channel for AnotherMockResource */
	type anotherMockResourceListWithNamespace struct {
		list      AnotherMockResourceList
		namespace string
	}
	anotherMockResourceChan := make(chan anotherMockResourceListWithNamespace)

	var initialAnotherMockResourceList AnotherMockResourceList
	/* Create channel for ClusterResource */
	/* Create channel for MockCustomType */
	type mockCustomTypeListWithNamespace struct {
		list      MockCustomTypeList
		namespace string
	}
	mockCustomTypeChan := make(chan mockCustomTypeListWithNamespace)

	var initialMockCustomTypeList MockCustomTypeList
	/* Create channel for MockCustomSpecHashType */
	type mockCustomSpecHashTypeListWithNamespace struct {
		list      MockCustomSpecHashTypeList
		namespace string
	}
	mockCustomSpecHashTypeChan := make(chan mockCustomSpecHashTypeListWithNamespace)

	var initialMockCustomSpecHashTypeList MockCustomSpecHashTypeList
	/* 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)

	var initialPodList github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodList

	currentSnapshot := TestingSnapshot{}
	simplemocksByNamespace := make(map[string]SimpleMockResourceList)
	mocksByNamespace := make(map[string]MockResourceList)
	fakesByNamespace := make(map[string]FakeResourceList)
	anothermockresourcesByNamespace := make(map[string]AnotherMockResourceList)
	mctsByNamespace := make(map[string]MockCustomTypeList)
	mcshtsByNamespace := make(map[string]MockCustomSpecHashTypeList)
	podsByNamespace := make(map[string]github_com_solo_io_solo_kit_pkg_api_v1_resources_common_kubernetes.PodList)

	for _, namespace := range watchNamespaces {
		/* Setup namespaced watch for SimpleMockResource */
		{
			simplemocks, err := c.simpleMockResource.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial SimpleMockResource list")
			}
			initialSimpleMockResourceList = append(initialSimpleMockResourceList, simplemocks...)
			simplemocksByNamespace[namespace] = simplemocks
		}
		simpleMockResourceNamespacesChan, simpleMockResourceErrs, err := c.simpleMockResource.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting SimpleMockResource watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, simpleMockResourceErrs, namespace+"-simplemocks")
		}(namespace)
		/* Setup namespaced watch for MockResource */
		{
			mocks, err := c.mockResource.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial MockResource list")
			}
			initialMockResourceList = append(initialMockResourceList, mocks...)
			mocksByNamespace[namespace] = mocks
		}
		mockResourceNamespacesChan, mockResourceErrs, err := c.mockResource.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting MockResource watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, mockResourceErrs, namespace+"-mocks")
		}(namespace)
		/* Setup namespaced watch for FakeResource */
		{
			fakes, err := c.fakeResource.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial FakeResource list")
			}
			initialFakeResourceList = append(initialFakeResourceList, fakes...)
			fakesByNamespace[namespace] = fakes
		}
		fakeResourceNamespacesChan, fakeResourceErrs, err := c.fakeResource.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting FakeResource watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, fakeResourceErrs, namespace+"-fakes")
		}(namespace)
		/* Setup namespaced watch for AnotherMockResource */
		{
			anothermockresources, err := c.anotherMockResource.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial AnotherMockResource list")
			}
			initialAnotherMockResourceList = append(initialAnotherMockResourceList, anothermockresources...)
			anothermockresourcesByNamespace[namespace] = anothermockresources
		}
		anotherMockResourceNamespacesChan, anotherMockResourceErrs, err := c.anotherMockResource.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting AnotherMockResource watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, anotherMockResourceErrs, namespace+"-anothermockresources")
		}(namespace)
		/* Setup namespaced watch for MockCustomType */
		{
			mcts, err := c.mockCustomType.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial MockCustomType list")
			}
			initialMockCustomTypeList = append(initialMockCustomTypeList, mcts...)
			mctsByNamespace[namespace] = mcts
		}
		mockCustomTypeNamespacesChan, mockCustomTypeErrs, err := c.mockCustomType.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting MockCustomType watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, mockCustomTypeErrs, namespace+"-mcts")
		}(namespace)
		/* Setup namespaced watch for MockCustomSpecHashType */
		{
			mcshts, err := c.mockCustomSpecHashType.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial MockCustomSpecHashType list")
			}
			initialMockCustomSpecHashTypeList = append(initialMockCustomSpecHashTypeList, mcshts...)
			mcshtsByNamespace[namespace] = mcshts
		}
		mockCustomSpecHashTypeNamespacesChan, mockCustomSpecHashTypeErrs, err := c.mockCustomSpecHashType.Watch(namespace, opts)
		if err != nil {
			return nil, nil, errors.Wrapf(err, "starting MockCustomSpecHashType watch")
		}

		done.Add(1)
		go func(namespace string) {
			defer done.Done()
			errutils.AggregateErrs(ctx, errs, mockCustomSpecHashTypeErrs, namespace+"-mcshts")
		}(namespace)
		/* Setup namespaced watch for Pod */
		{
			pods, err := c.pod.List(namespace, clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
			if err != nil {
				return nil, nil, errors.Wrapf(err, "initial Pod list")
			}
			initialPodList = append(initialPodList, pods...)
			podsByNamespace[namespace] = pods
		}
		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)

		/* Watch for changes and update snapshot */
		go func(namespace string) {
			for {
				select {
				case <-ctx.Done():
					return
				case simpleMockResourceList, ok := <-simpleMockResourceNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case simpleMockResourceChan <- simpleMockResourceListWithNamespace{list: simpleMockResourceList, namespace: namespace}:
					}
				case mockResourceList, ok := <-mockResourceNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case mockResourceChan <- mockResourceListWithNamespace{list: mockResourceList, namespace: namespace}:
					}
				case fakeResourceList, ok := <-fakeResourceNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case fakeResourceChan <- fakeResourceListWithNamespace{list: fakeResourceList, namespace: namespace}:
					}
				case anotherMockResourceList, ok := <-anotherMockResourceNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case anotherMockResourceChan <- anotherMockResourceListWithNamespace{list: anotherMockResourceList, namespace: namespace}:
					}
				case mockCustomTypeList, ok := <-mockCustomTypeNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case mockCustomTypeChan <- mockCustomTypeListWithNamespace{list: mockCustomTypeList, namespace: namespace}:
					}
				case mockCustomSpecHashTypeList, ok := <-mockCustomSpecHashTypeNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case mockCustomSpecHashTypeChan <- mockCustomSpecHashTypeListWithNamespace{list: mockCustomSpecHashTypeList, namespace: namespace}:
					}
				case podList, ok := <-podNamespacesChan:
					if !ok {
						return
					}
					select {
					case <-ctx.Done():
						return
					case podChan <- podListWithNamespace{list: podList, namespace: namespace}:
					}
				}
			}
		}(namespace)
	}
	/* Initialize snapshot for Simplemocks */
	currentSnapshot.Simplemocks = initialSimpleMockResourceList.Sort()
	/* Initialize snapshot for Mocks */
	currentSnapshot.Mocks = initialMockResourceList.Sort()
	/* Initialize snapshot for Fakes */
	currentSnapshot.Fakes = initialFakeResourceList.Sort()
	/* Initialize snapshot for Anothermockresources */
	currentSnapshot.Anothermockresources = initialAnotherMockResourceList.Sort()
	/* Setup cluster-wide watch for ClusterResource */
	var err error
	currentSnapshot.Clusterresources, err = c.clusterResource.List(clients.ListOpts{Ctx: opts.Ctx, Selector: opts.Selector})
	if err != nil {
		return nil, nil, errors.Wrapf(err, "initial ClusterResource list")
	}
	clusterResourceChan, clusterResourceErrs, err := c.clusterResource.Watch(opts)
	if err != nil {
		return nil, nil, errors.Wrapf(err, "starting ClusterResource watch")
	}
	done.Add(1)
	go func() {
		defer done.Done()
		errutils.AggregateErrs(ctx, errs, clusterResourceErrs, "clusterresources")
	}()
	/* Initialize snapshot for Mcts */
	currentSnapshot.Mcts = initialMockCustomTypeList.Sort()
	/* Initialize snapshot for Mcshts */
	currentSnapshot.Mcshts = initialMockCustomSpecHashTypeList.Sort()
	/* Initialize snapshot for Pods */
	currentSnapshot.Pods = initialPodList.Sort()

	snapshots := make(chan *TestingSnapshot)
	go func() {
		// sent initial snapshot to kick off the watch
		initialSnapshot := currentSnapshot.Clone()
		snapshots <- &initialSnapshot

		timer := time.NewTicker(time.Second * 1)
		previousHash, err := currentSnapshot.Hash(nil)
		if err != nil {
			contextutils.LoggerFrom(ctx).Panicw("error while hashing, this should never happen", zap.Error(err))
		}
		sync := func() {
			currentHash, err := currentSnapshot.Hash(nil)
			// this should never happen, so panic if it does
			if err != nil {
				contextutils.LoggerFrom(ctx).Panicw("error while hashing, this should never happen", zap.Error(err))
			}
			if previousHash == currentHash {
				return
			}

			sentSnapshot := currentSnapshot.Clone()
			select {
			case snapshots <- &sentSnapshot:
				stats.Record(ctx, mTestingSnapshotOut.M(1))
				previousHash = currentHash
			default:
				stats.Record(ctx, mTestingSnapshotMissed.M(1))
			}
		}

		defer func() {
			close(snapshots)
			// we must wait for done before closing the error chan,
			// to avoid sending on close channel.
			done.Wait()
			close(errs)
		}()
		for {
			record := func() { stats.Record(ctx, mTestingSnapshotIn.M(1)) }

			select {
			case <-timer.C:
				sync()
			case <-ctx.Done():
				return
			case <-c.forceEmit:
				sentSnapshot := currentSnapshot.Clone()
				snapshots <- &sentSnapshot
			case simpleMockResourceNamespacedList, ok := <-simpleMockResourceChan:
				if !ok {
					return
				}
				record()

				namespace := simpleMockResourceNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"simple_mock_resource",
					mTestingResourcesIn,
				)

				// merge lists by namespace
				simplemocksByNamespace[namespace] = simpleMockResourceNamespacedList.list
				var simpleMockResourceList SimpleMockResourceList
				for _, simplemocks := range simplemocksByNamespace {
					simpleMockResourceList = append(simpleMockResourceList, simplemocks...)
				}
				currentSnapshot.Simplemocks = simpleMockResourceList.Sort()
			case mockResourceNamespacedList, ok := <-mockResourceChan:
				if !ok {
					return
				}
				record()

				namespace := mockResourceNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"mock_resource",
					mTestingResourcesIn,
				)

				// merge lists by namespace
				mocksByNamespace[namespace] = mockResourceNamespacedList.list
				var mockResourceList MockResourceList
				for _, mocks := range mocksByNamespace {
					mockResourceList = append(mockResourceList, mocks...)
				}
				currentSnapshot.Mocks = mockResourceList.Sort()
			case fakeResourceNamespacedList, ok := <-fakeResourceChan:
				if !ok {
					return
				}
				record()

				namespace := fakeResourceNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"fake_resource",
					mTestingResourcesIn,
				)

				// merge lists by namespace
				fakesByNamespace[namespace] = fakeResourceNamespacedList.list
				var fakeResourceList FakeResourceList
				for _, fakes := range fakesByNamespace {
					fakeResourceList = append(fakeResourceList, fakes...)
				}
				currentSnapshot.Fakes = fakeResourceList.Sort()
			case anotherMockResourceNamespacedList, ok := <-anotherMockResourceChan:
				if !ok {
					return
				}
				record()

				namespace := anotherMockResourceNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"another_mock_resource",
					mTestingResourcesIn,
				)

				// merge lists by namespace
				anothermockresourcesByNamespace[namespace] = anotherMockResourceNamespacedList.list
				var anotherMockResourceList AnotherMockResourceList
				for _, anothermockresources := range anothermockresourcesByNamespace {
					anotherMockResourceList = append(anotherMockResourceList, anothermockresources...)
				}
				currentSnapshot.Anothermockresources = anotherMockResourceList.Sort()
			case clusterResourceList, ok := <-clusterResourceChan:
				if !ok {
					return
				}
				record()

				skstats.IncrementResourceCount(
					ctx,
					"<all>",
					"cluster_resource",
					mTestingResourcesIn,
				)

				currentSnapshot.Clusterresources = clusterResourceList
			case mockCustomTypeNamespacedList, ok := <-mockCustomTypeChan:
				if !ok {
					return
				}
				record()

				namespace := mockCustomTypeNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"mock_custom_type",
					mTestingResourcesIn,
				)

				// merge lists by namespace
				mctsByNamespace[namespace] = mockCustomTypeNamespacedList.list
				var mockCustomTypeList MockCustomTypeList
				for _, mcts := range mctsByNamespace {
					mockCustomTypeList = append(mockCustomTypeList, mcts...)
				}
				currentSnapshot.Mcts = mockCustomTypeList.Sort()
			case mockCustomSpecHashTypeNamespacedList, ok := <-mockCustomSpecHashTypeChan:
				if !ok {
					return
				}
				record()

				namespace := mockCustomSpecHashTypeNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"mock_custom_spec_hash_type",
					mTestingResourcesIn,
				)

				// merge lists by namespace
				mcshtsByNamespace[namespace] = mockCustomSpecHashTypeNamespacedList.list
				var mockCustomSpecHashTypeList MockCustomSpecHashTypeList
				for _, mcshts := range mcshtsByNamespace {
					mockCustomSpecHashTypeList = append(mockCustomSpecHashTypeList, mcshts...)
				}
				currentSnapshot.Mcshts = mockCustomSpecHashTypeList.Sort()
			case podNamespacedList, ok := <-podChan:
				if !ok {
					return
				}
				record()

				namespace := podNamespacedList.namespace

				skstats.IncrementResourceCount(
					ctx,
					namespace,
					"pod",
					mTestingResourcesIn,
				)

				// 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()
			}
		}
	}()
	return snapshots, errs, nil
}