func expectedIngressPath(host, path, serviceName string) string {
	rulePath := path
	if host != "" {
		rulePath = host + path
	}
	if rulePath == "" {
		rulePath = `"" (default path)`
	} else {
		rulePath = fmt.Sprintf("%q", rulePath)
	}
	return fmt.Sprintf("%s -> %q", rulePath, serviceName)
}
expectedIngressPath is a helper to print a useful error message.
package await
import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"time"
	"github.com/pkg/errors"
	"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/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"
	networkingv1b1 "k8s.io/api/networking/v1beta1"
	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/runtime/schema"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/dynamic"
)
const (
	DefaultIngressTimeoutMins = 10
)
type ingressInitAwaiter struct {
	config                    createAwaitConfig
	ingress                   *unstructured.Unstructured
	ingressReady              bool
	endpointsSettled          bool
	knownEndpointObjects      sets.String
	knownExternalNameServices sets.String
}
func makeIngressInitAwaiter(c createAwaitConfig) *ingressInitAwaiter {
	return &ingressInitAwaiter{
		config:                    c,
		ingress:                   c.currentOutputs,
		ingressReady:              false,
		endpointsSettled:          false,
		knownEndpointObjects:      sets.NewString(),
		knownExternalNameServices: sets.NewString(),
	}
}
func awaitIngressInit(c createAwaitConfig) error {
	return makeIngressInitAwaiter(c).Await()
}
func awaitIngressRead(c createAwaitConfig) error {
	return makeIngressInitAwaiter(c).Read()
}
func awaitIngressUpdate(u updateAwaitConfig) error {
	return makeIngressInitAwaiter(u.createAwaitConfig).Await()
}
func (iia *ingressInitAwaiter) Await() error {
	stopper := make(chan struct{})
	defer close(stopper)
	informerFactory := informers.NewInformerFactory(iia.config.clientSet,
		informers.WithNamespaceOrDefault(iia.config.currentInputs.GetNamespace()))
	informerFactory.Start(stopper)
	ingressEvents := make(chan watch.Event)
	ingressInformer, err := informers.New(informerFactory, informers.ForGVR(schema.GroupVersionResource{
		Group:    "networking.k8s.io",
		Version:  "v1",
		Resource: "ingresses",
	}), informers.WithEventChannel(ingressEvents))
	if err != nil {
		return err
	}
	go ingressInformer.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)
	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)
	timeout := metadata.TimeoutDuration(iia.config.timeout, iia.config.currentInputs, DefaultIngressTimeoutMins*60)
	return iia.await(ingressEvents, serviceEvents, endpointsEvents, make(chan struct{}), time.After(timeout))
}
func (iia *ingressInitAwaiter) Read() error {
	ingressClient, endpointsClient, servicesClient, err := iia.makeClients()
	if err != nil {
		return err
	}
	ingress, err := ingressClient.Get(context.TODO(), iia.config.currentInputs.GetName(), metav1.GetOptions{})
	if err != nil {
		return err
	}
	endpointList, err := endpointsClient.List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		logger.V(3).Infof("Failed to list endpoints needed for Ingress awaiter: %v", err)
		endpointList = &unstructured.UnstructuredList{Items: []unstructured.Unstructured{}}
	}
	serviceList, err := servicesClient.List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		logger.V(3).Infof("Failed to list services needed for Ingress awaiter: %v", err)
		serviceList = &unstructured.UnstructuredList{Items: []unstructured.Unstructured{}}
	}
	return iia.read(ingress, endpointList, serviceList)
}
func (iia *ingressInitAwaiter) read(ingress *unstructured.Unstructured, endpoints *unstructured.UnstructuredList,
	services *unstructured.UnstructuredList) error {
	iia.processIngressEvent(watchAddedEvent(ingress))
	err := services.EachListItem(func(service runtime.Object) error {
		iia.processServiceEvent(watchAddedEvent(service.(*unstructured.Unstructured)))
		return nil
	})
	if err != nil {
		logger.V(3).Infof("Error iterating over endpoint list for service %q: %v", ingress.GetName(), err)
	}
	settled := make(chan struct{})
	logger.V(3).Infof("Processing endpoint list: %#v", endpoints)
	err = endpoints.EachListItem(func(endpoint runtime.Object) error {
		iia.processEndpointEvent(watchAddedEvent(endpoint.(*unstructured.Unstructured)), settled)
		return nil
	})
	if err != nil {
		logger.V(3).Infof("Error iterating over endpoint list for ingress %q: %v", ingress.GetName(), err)
	}
	iia.endpointsSettled = true
	if iia.checkAndLogStatus() {
		return nil
	}
	return &initializationError{
		subErrors: iia.errorMessages(),
		object:    ingress,
	}
}
func (iia *ingressInitAwaiter) await(
	ingressEvents, serviceEvents, endpointsEvents <-chan watch.Event,
	settled chan struct{},
	timeout <-chan time.Time,
) error {
	iia.config.logStatus(diag.Info, "[1/3] Finding a matching service for each Ingress path")
	for {
		if iia.checkAndLogStatus() {
			return nil
		}
		select {
		case <-iia.config.ctx.Done():
			if iia.ingressReady && iia.checkIfEndpointsReady() {
				return nil
			}
			return &cancellationError{
				object:    iia.ingress,
				subErrors: iia.errorMessages(),
			}
		case <-timeout:
			if iia.ingressReady && iia.checkIfEndpointsReady() {
				return nil
			}
			return &timeoutError{
				object:    iia.ingress,
				subErrors: iia.errorMessages(),
			}
		case <-settled:
			iia.endpointsSettled = true
		case event := <-ingressEvents:
			iia.processIngressEvent(event)
		case event := <-endpointsEvents:
			iia.processEndpointEvent(event, settled)
		case event := <-serviceEvents:
			iia.processServiceEvent(event)
		}
	}
}
func (iia *ingressInitAwaiter) processServiceEvent(event watch.Event) {
	service, isUnstructured := event.Object.(*unstructured.Unstructured)
	if !isUnstructured {
		logger.V(3).Infof("Service watch received unknown object type %q",
			reflect.TypeOf(service))
		return
	}
	name := service.GetName()
	if event.Type == watch.Deleted {
		iia.knownExternalNameServices.Delete(name)
		return
	}
	t, ok := openapi.Pluck(service.Object, "spec", "type")
	if ok && t.(string) == "ExternalName" {
		iia.knownExternalNameServices.Insert(name)
	}
}
func (iia *ingressInitAwaiter) processIngressEvent(event watch.Event) {
	inputIngressName := iia.config.currentInputs.GetName()
	ingress, isUnstructured := event.Object.(*unstructured.Unstructured)
	if !isUnstructured {
		logger.V(3).Infof("Ingress watch received unknown object type %q",
			reflect.TypeOf(ingress))
		return
	}
	if ingress.GetName() != inputIngressName {
		return
	}
	iia.ingressReady = false
	if event.Type == watch.Deleted {
		return
	}
	iia.ingress = ingress
	obj, err := decodeIngress(ingress)
	if err != nil {
		logger.V(3).Infof("Unable to decode Ingress object from unstructured: %#v", ingress)
		return
	}
	logger.V(3).Infof("Received status for ingress %q: %#v", inputIngressName, obj.Status)
	iia.ingressReady = len(obj.Status.LoadBalancer.Ingress) > 0
	logger.V(3).Infof("Waiting for ingress %q to update .status.loadBalancer with hostname/IP",
		inputIngressName)
}
func decodeIngress(u *unstructured.Unstructured) (*networkingv1b1.Ingress, error) {
	b, err := u.MarshalJSON()
	if err != nil {
		return nil, err
	}
	var obj networkingv1b1.Ingress
	err = json.Unmarshal(b, &obj)
	if err != nil {
		return nil, err
	}
	return &obj, nil
}
func (iia *ingressInitAwaiter) checkIfEndpointsReady() bool {
	obj, err := decodeIngress(iia.ingress)
	if err != nil {
		logger.V(3).Infof("Unable to decode Ingress object from unstructured: %#v", iia.ingress)
		return false
	}
	for _, rule := range obj.Spec.Rules {
		if rule.HTTP == nil {
			iia.config.logStatus(diag.Error, fmt.Sprintf("expected value %q is unset for ingress: %s",
				".spec.rules[*].http", obj.Name))
			return false
		}
		for _, path := range rule.HTTP.Paths {
			if iia.knownExternalNameServices.Has(path.Backend.ServiceName) {
				continue
			}
			if !iia.knownEndpointObjects.Has(path.Backend.ServiceName) {
				iia.config.logStatus(diag.Info, fmt.Sprintf("No matching service found for ingress rule: %s",
					expectedIngressPath(rule.Host, path.Path, path.Backend.ServiceName)))
				return false
			}
		}
	}
	return true
}