func (sia *serviceInitAwaiter) await(
	serviceEvents,
	endpointsEvents <-chan watch.Event,
	timeout <-chan time.Time,
	settled chan struct{},
	version cluster.ServerVersion,
) error {
	sia.config.logStatus(diag.Info, "[1/3] Finding Pods to direct traffic to")
	for {
		if sia.checkAndLogStatus(version) {
			return nil
		}
		select {
		case <-sia.config.ctx.Done():
			if sia.serviceReady && sia.endpointsReady {
				return nil
			}
			return &cancellationError{
				object:    sia.service,
				subErrors: sia.errorMessages(),
			}
		case <-timeout:
			if sia.serviceReady && sia.endpointsReady {
				return nil
			}
			return &timeoutError{
				object:    sia.service,
				subErrors: sia.errorMessages(),
			}
		case <-settled:
			sia.endpointsSettled = true
		case event := <-serviceEvents:
			sia.processServiceEvent(event)
		case event := <-endpointsEvents:
			sia.processEndpointEvent(event, settled)
		}
	}
}
await is a helper companion to `Await` designed to make it easy to test this module.
package await
import (
	"context"
	"fmt"
	"reflect"
	"time"
	"github.com/pkg/errors"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/dynamic"
	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/await/informers"
	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/clients"
	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/cluster"
	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/kinds"
	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/metadata"
	"github.com/pulumi/pulumi-kubernetes/provider/v3/pkg/openapi"
	"github.com/pulumi/pulumi/sdk/v3/go/common/diag"
	"github.com/pulumi/pulumi/sdk/v3/go/common/util/cmdutil"
	logger "github.com/pulumi/pulumi/sdk/v3/go/common/util/logging"
)
const (
	DefaultServiceTimeoutMins = 10
)
type serviceInitAwaiter struct {
	config           createAwaitConfig
	service          *unstructured.Unstructured
	serviceReady     bool
	endpointsReady   bool
	endpointsSettled bool
	serviceType      string
}
func makeServiceInitAwaiter(c createAwaitConfig) *serviceInitAwaiter {
	specType, _ := openapi.Pluck(c.currentOutputs.Object, "spec", "type")
	var t string
	if specTypeString, isString := specType.(string); isString {
		t = specTypeString
	} else {
		t = string(v1.ServiceTypeClusterIP)
	}
	return &serviceInitAwaiter{
		config:           c,
		service:          c.currentOutputs,
		serviceReady:     false,
		endpointsReady:   false,
		endpointsSettled: false,
		serviceType:      t,
	}
}
func awaitServiceInit(c createAwaitConfig) error {
	return makeServiceInitAwaiter(c).Await()
}
func awaitServiceRead(c createAwaitConfig) error {
	return makeServiceInitAwaiter(c).Read()
}
func awaitServiceUpdate(u updateAwaitConfig) error {
	return makeServiceInitAwaiter(u.createAwaitConfig).Await()
}
func (sia *serviceInitAwaiter) Await() error {
	stopper := make(chan struct{})
	defer close(stopper)
	informerFactory := informers.NewInformerFactory(sia.config.clientSet,
		informers.WithNamespaceOrDefault(sia.config.currentInputs.GetNamespace()))
	informerFactory.Start(stopper)
	serviceEvents := make(chan watch.Event)
	serviceInformer, err := informers.New(informerFactory, informers.ForServices(), informers.WithEventChannel(serviceEvents))
	if err != nil {
		return err
	}
	go serviceInformer.Informer().Run(stopper)
	endpointsEvents := make(chan watch.Event)
	endpointsInformer, err := informers.New(informerFactory, informers.ForEndpoints(), informers.WithEventChannel(endpointsEvents))
	if err != nil {
		return err
	}
	go endpointsInformer.Informer().Run(stopper)
	version := cluster.TryGetServerVersion(sia.config.clientSet.DiscoveryClientCached)
	timeout := metadata.TimeoutDuration(sia.config.timeout, sia.config.currentInputs, DefaultServiceTimeoutMins*60)
	return sia.await(serviceEvents, endpointsEvents, time.After(timeout), make(chan struct{}), version)
}
func (sia *serviceInitAwaiter) Read() error {
	serviceClient, endpointsClient, err := sia.makeClients()
	if err != nil {
		return err
	}
	service, err := serviceClient.Get(context.TODO(),
		sia.config.currentOutputs.GetName(),
		metav1.GetOptions{})
	if err != nil {
		return err
	}
	endpointList, err := endpointsClient.List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		logger.V(3).Infof("Error retrieving ReplicaSet list for Service %q: %v",
			service.GetName(), err)
		endpointList = &unstructured.UnstructuredList{Items: []unstructured.Unstructured{}}
	}
	version := cluster.TryGetServerVersion(sia.config.clientSet.DiscoveryClientCached)
	return sia.read(service, endpointList, version)
}
func (sia *serviceInitAwaiter) read(
	service *unstructured.Unstructured, endpoints *unstructured.UnstructuredList,
	version cluster.ServerVersion,
) error {
	sia.processServiceEvent(watchAddedEvent(service))
	var err error
	settled := make(chan struct{})
	logger.V(3).Infof("Processing endpoint list: %#v", endpoints)
	err = endpoints.EachListItem(func(endpoint runtime.Object) error {
		sia.processEndpointEvent(watchAddedEvent(endpoint.(*unstructured.Unstructured)), settled)
		return nil
	})
	if err != nil {
		logger.V(3).Infof("Error iterating over ReplicaSet list for Deployment %q: %v",
			service.GetName(), err)
	}
	sia.endpointsSettled = true
	if sia.checkAndLogStatus(version) {
		return nil
	}
	return &initializationError{
		subErrors: sia.errorMessages(),
		object:    service,
	}
}