// Copyright 2018 Istio Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package model

import (
	"encoding/json"
	"sort"
	"sync"
	"time"

	networking "istio.io/api/networking/v1alpha3"

	"istio.io/istio/pilot/pkg/features"
	"istio.io/istio/pilot/pkg/monitoring"
	"istio.io/istio/pkg/config/constants"
	"istio.io/istio/pkg/config/host"
	"istio.io/istio/pkg/config/labels"
	"istio.io/istio/pkg/config/protocol"
	"istio.io/istio/pkg/config/visibility"
)

// PushContext tracks the status of a push - metrics and errors.
// Metrics are reset after a push - at the beginning all
// values are zero, and when push completes the status is reset.
// The struct is exposed in a debug endpoint - fields public to allow
// easy serialization as json.
type PushContext struct {
	proxyStatusMutex sync.RWMutex
	// ProxyStatus is keyed by the error code, and holds a map keyed
	// by the ID.
	ProxyStatus map[string]map[string]ProxyPushStatus

	// Mutex is used to protect the below store.
	// All data is set when the PushContext object is populated in `InitContext`,
	// data should not be changed by plugins.
	Mutex sync.Mutex `json:"-"`

	// Synthesized from env.Mesh
	defaultServiceExportTo         map[visibility.Instance]bool
	defaultVirtualServiceExportTo  map[visibility.Instance]bool
	defaultDestinationRuleExportTo map[visibility.Instance]bool

	// privateServices are reachable within the same namespace.
	privateServicesByNamespace map[string][]*Service
	// publicServices are services reachable within the mesh.
	publicServices []*Service

	privateVirtualServicesByNamespace map[string][]Config
	publicVirtualServices             []Config

	// destination rules are of three types:
	//  namespaceLocalDestRules: all public/private dest rules pertaining to a service defined in a given namespace
	//  namespaceExportedDestRules: all public dest rules pertaining to a service defined in a namespace
	//  allExportedDestRules: all (public) dest rules across all namespaces
	// We need the allExportedDestRules in addition to namespaceExportedDestRules because we select
	// the dest rule based on the most specific host match, and not just any destination rule
	namespaceLocalDestRules    map[string]*processedDestRules
	namespaceExportedDestRules map[string]*processedDestRules
	allExportedDestRules       *processedDestRules

	// sidecars for each namespace
	sidecarsByNamespace map[string][]*SidecarScope
	// envoy filters for each namespace including global config namespace
	envoyFiltersByNamespace map[string][]*EnvoyFilterWrapper
	// gateways for each namespace
	gatewaysByNamespace map[string][]Config
	allGateways         []Config
	////////// END ////////

	// The following data is either a global index or used in the inbound path.
	// Namespace specific views do not apply here.

	// ServiceByHostnameAndNamespace has all services, indexed by hostname then namesace.
	ServiceByHostnameAndNamespace map[host.Name]map[string]*Service `json:"-"`

	// AuthzPolicies stores the existing authorization policies in the cluster. Could be nil if there
	// are no authorization policies in the cluster.
	AuthzPolicies *AuthorizationPolicies `json:"-"`

	// Env has a pointer to the shared environment used to create the snapshot.
	Env *Environment `json:"-"`

	// ServiceAccounts contains a map of hostname and port to service accounts.
	ServiceAccounts map[host.Name]map[int][]string `json:"-"`

	initDone bool
}

type processedDestRules struct {
	// List of dest rule hosts. We match with the most specific host first
	hosts []host.Name
	// Map of dest rule host and the merged destination rules for that host
	destRule map[host.Name]*combinedDestinationRule
}

// XDSUpdater is used for direct updates of the xDS model and incremental push.
// Pilot uses multiple registries - for example each K8S cluster is a registry instance,
// as well as consul and future EDS or MCP sources. Each registry is responsible for
// tracking a set of endpoints associated with mesh services, and calling the EDSUpdate
// on changes. A registry may group endpoints for a service in smaller subsets - for
// example by deployment, or to deal with very large number of endpoints for a service.
// We want to avoid passing around large objects - like full list of endpoints for a registry,
// or the full list of endpoints for a service across registries, since it limits scalability.
//
// Future optimizations will include grouping the endpoints by labels, gateway or region to
// reduce the time when subsetting or split-horizon is used. This design assumes pilot
// tracks all endpoints in the mesh and they fit in RAM - so limit is few M endpoints.
// It is possible to split the endpoint tracking in future.
type XDSUpdater interface {

	// EDSUpdate is called when the list of endpoints or labels in a ServiceEntry is
	// changed. For each cluster and hostname, the full list of active endpoints (including empty list)
	// must be sent. The shard name is used as a key - current implementation is using the registry
	// name.
	EDSUpdate(shard, hostname string, namespace string, entry []*IstioEndpoint) error

	// SvcUpdate is called when a service port mapping definition is updated.
	// This interface is WIP - labels, annotations and other changes to service may be
	// updated to force a EDS and CDS recomputation and incremental push, as it doesn't affect
	// LDS/RDS.
	SvcUpdate(shard, hostname string, ports map[string]uint32, rports map[uint32]string)

	// WorkloadUpdate is called by a registry when the labels or annotations on a workload have changed.
	// The 'id' is the IP address of the pod for k8s if the pod is in the main/default network.
	// In future it will include the 'network id' for pods in a different network, behind a zvpn gate.
	// The IP is used because K8S Endpoints object associated with a Service only include the IP.
	// We use Endpoints to track the membership to a service and readiness.
	WorkloadUpdate(id string, labels map[string]string, annotations map[string]string)

	// ConfigUpdate is called to notify the XDS server of config updates and request a push.
	// The requests may be collapsed and throttled.
	// This replaces the 'cache invalidation' model.
	ConfigUpdate(req *PushRequest)
}

// PushRequest defines a request to push to proxies
// It is used to send updates to the config update debouncer and pass to the PushQueue.
type PushRequest struct {
	// Full determines whether a full push is required or not. If set to false, only endpoints will be sent.
	Full bool

	// TargetNamespaces contains a list of namespaces that were changed in the update.
	// This is used as an optimization to avoid unnecessary pushes to proxies that are scoped with a Sidecar.
	// Currently, this will only scope EDS updates, as config updates are more complicated.
	// If this is empty, then proxies in all namespaces will get an update
	// If this is present, then only proxies that import this namespace will get an update
	TargetNamespaces map[string]struct{}

	// EdsUpdates keeps track of all service updated since last full push.
	// Key is the hostname (serviceName).
	// This is used by incremental eds.
	EdsUpdates map[string]struct{}

	// Push stores the push context to use for the update. This may initially be nil, as we will
	// debounce changes before a PushContext is eventually created.
	Push *PushContext

	// Start represents the time a push was started. This represents the time of adding to the PushQueue.
	// Note that this does not include time spent debouncing.
	Start time.Time
}

// Merge two update requests together
func (first *PushRequest) Merge(other *PushRequest) *PushRequest {
	if first == nil {
		return other
	}
	if other == nil {
		return first
	}

	merged := &PushRequest{
		// Keep the first (older) start time
		Start: first.Start,

		// If either is full we need a full push
		Full: first.Full || other.Full,

		// The other push context is presumed to be later and more up to date
		Push: other.Push,
	}

	// Only merge EdsUpdates when incremental eds push needed.
	if !merged.Full {
		merged.EdsUpdates = make(map[string]struct{})
		// Merge the updates
		for update := range first.EdsUpdates {
			merged.EdsUpdates[update] = struct{}{}
		}
		for update := range other.EdsUpdates {
			merged.EdsUpdates[update] = struct{}{}
		}
	} else {
		merged.EdsUpdates = nil
	}

	if !features.ScopePushes.Get() {
		// If push scoping is not enabled, we do not care about target namespaces
		return merged
	}

	// If either does not specify only namespaces, this means update all namespaces
	if len(first.TargetNamespaces) == 0 || len(other.TargetNamespaces) == 0 {
		return merged
	}

	// Merge the target namespaces
	merged.TargetNamespaces = make(map[string]struct{})
	for update := range first.TargetNamespaces {
		merged.TargetNamespaces[update] = struct{}{}
	}
	for update := range other.TargetNamespaces {
		merged.TargetNamespaces[update] = struct{}{}
	}

	return merged
}

// ProxyPushStatus represents an event captured during config push to proxies.
// It may contain additional message and the affected proxy.
type ProxyPushStatus struct {
	Proxy   string `json:"proxy,omitempty"`
	Message string `json:"message,omitempty"`
}

type combinedDestinationRule struct {
	subsets map[string]struct{} // list of subsets seen so far
	// We are not doing ports
	config *Config
}

// Add will add an case to the metric.
func (ps *PushContext) Add(metric monitoring.Metric, key string, proxy *Proxy, msg string) {
	if ps == nil {
		log.Infof("Metric without context %s %v %s", key, proxy, msg)
		return
	}
	ps.proxyStatusMutex.Lock()
	defer ps.proxyStatusMutex.Unlock()

	metricMap, f := ps.ProxyStatus[metric.Name()]
	if !f {
		metricMap = map[string]ProxyPushStatus{}
		ps.ProxyStatus[metric.Name()] = metricMap
	}
	ev := ProxyPushStatus{Message: msg}
	if proxy != nil {
		ev.Proxy = proxy.ID
	}
	metricMap[key] = ev
}

var (

	// EndpointNoPod tracks endpoints without an associated pod. This is an error condition, since
	// we can't figure out the labels. It may be a transient problem, if endpoint is processed before
	// pod.
	EndpointNoPod = monitoring.NewGauge(
		"endpoint_no_pod",
		"Endpoints without an associated pod.",
	)

	// ProxyStatusNoService represents proxies not selected by any service
	// This can be normal - for workloads that act only as client, or are not covered by a Service.
	// It can also be an error, for example in cases the Endpoint list of a service was not updated by the time
	// the sidecar calls.
	// Updated by GetProxyServiceInstances
	ProxyStatusNoService = monitoring.NewGauge(
		"pilot_no_ip",
		"Pods not found in the endpoint table, possibly invalid.",
	)

	// ProxyStatusEndpointNotReady represents proxies found not be ready.
	// Updated by GetProxyServiceInstances. Normal condition when starting
	// an app with readiness, error if it doesn't change to 0.
	ProxyStatusEndpointNotReady = monitoring.NewGauge(
		"pilot_endpoint_not_ready",
		"Endpoint found in unready state.",
	)

	// ProxyStatusConflictOutboundListenerTCPOverHTTP metric tracks number of
	// wildcard TCP listeners that conflicted with existing wildcard HTTP listener on same port
	ProxyStatusConflictOutboundListenerTCPOverHTTP = monitoring.NewGauge(
		"pilot_conflict_outbound_listener_tcp_over_current_http",
		"Number of conflicting wildcard tcp listeners with current wildcard http listener.",
	)

	// ProxyStatusConflictOutboundListenerHTTPoverHTTPS metric tracks number of
	// HTTP listeners that conflicted with well known HTTPS ports
	ProxyStatusConflictOutboundListenerHTTPoverHTTPS = monitoring.NewGauge(
		"pilot_conflict_outbound_listener_http_over_https",
		"Number of conflicting HTTP listeners with well known HTTPS ports",
	)

	// ProxyStatusConflictOutboundListenerTCPOverTCP metric tracks number of
	// TCP listeners that conflicted with existing TCP listeners on same port
	ProxyStatusConflictOutboundListenerTCPOverTCP = monitoring.NewGauge(
		"pilot_conflict_outbound_listener_tcp_over_current_tcp",
		"Number of conflicting tcp listeners with current tcp listener.",
	)

	// ProxyStatusConflictOutboundListenerHTTPOverTCP metric tracks number of
	// wildcard HTTP listeners that conflicted with existing wildcard TCP listener on same port
	ProxyStatusConflictOutboundListenerHTTPOverTCP = monitoring.NewGauge(
		"pilot_conflict_outbound_listener_http_over_current_tcp",
		"Number of conflicting wildcard http listeners with current wildcard tcp listener.",
	)

	// ProxyStatusConflictInboundListener tracks cases of multiple inbound
	// listeners - 2 services selecting the same port of the pod.
	ProxyStatusConflictInboundListener = monitoring.NewGauge(
		"pilot_conflict_inbound_listener",
		"Number of conflicting inbound listeners.",
	)

	// DuplicatedClusters tracks duplicate clusters seen while computing CDS
	DuplicatedClusters = monitoring.NewGauge(
		"pilot_duplicate_envoy_clusters",
		"Duplicate envoy clusters caused by service entries with same hostname",
	)

	// ProxyStatusClusterNoInstances tracks clusters (services) without workloads.
	ProxyStatusClusterNoInstances = monitoring.NewGauge(
		"pilot_eds_no_instances",
		"Number of clusters without instances.",
	)

	// DuplicatedDomains tracks rejected VirtualServices due to duplicated hostname.
	DuplicatedDomains = monitoring.NewGauge(
		"pilot_vservice_dup_domain",
		"Virtual services with dup domains.",
	)

	// DuplicatedSubsets tracks duplicate subsets that we rejected while merging multiple destination rules for same host
	DuplicatedSubsets = monitoring.NewGauge(
		"pilot_destrule_subsets",
		"Duplicate subsets across destination rules for same host",
	)

	// totalVirtualServices tracks the total number of virtual service
	totalVirtualServices = monitoring.NewGauge(
		"pilot_virt_services",
		"Total virtual services known to pilot.",
	)

	// LastPushStatus preserves the metrics and data collected during lasts global push.
	// It can be used by debugging tools to inspect the push event. It will be reset after each push with the
	// new version.
	LastPushStatus *PushContext
	// LastPushMutex will protect the LastPushStatus
	LastPushMutex sync.Mutex

	// All metrics we registered.
	metrics = []monitoring.Metric{
		EndpointNoPod,
		ProxyStatusNoService,
		ProxyStatusEndpointNotReady,
		ProxyStatusConflictOutboundListenerTCPOverHTTP,
		ProxyStatusConflictOutboundListenerHTTPoverHTTPS,
		ProxyStatusConflictOutboundListenerTCPOverTCP,
		ProxyStatusConflictOutboundListenerHTTPOverTCP,
		ProxyStatusConflictInboundListener,
		DuplicatedClusters,
		ProxyStatusClusterNoInstances,
		DuplicatedDomains,
		DuplicatedSubsets,
	}
)

func init() {
	for _, m := range metrics {
		monitoring.MustRegisterViews(m)
	}
	monitoring.MustRegisterViews(totalVirtualServices)
}

// NewPushContext creates a new PushContext structure to track push status.
func NewPushContext() *PushContext {
	// TODO: detect push in progress, don't update status if set
	return &PushContext{
		publicServices:                    []*Service{},
		privateServicesByNamespace:        map[string][]*Service{},
		publicVirtualServices:             []Config{},
		privateVirtualServicesByNamespace: map[string][]Config{},
		namespaceLocalDestRules:           map[string]*processedDestRules{},
		namespaceExportedDestRules:        map[string]*processedDestRules{},
		allExportedDestRules: &processedDestRules{
			hosts:    make([]host.Name, 0),
			destRule: map[host.Name]*combinedDestinationRule{},
		},
		sidecarsByNamespace:           map[string][]*SidecarScope{},
		envoyFiltersByNamespace:       map[string][]*EnvoyFilterWrapper{},
		gatewaysByNamespace:           map[string][]Config{},
		allGateways:                   []Config{},
		ServiceByHostnameAndNamespace: map[host.Name]map[string]*Service{},
		ProxyStatus:                   map[string]map[string]ProxyPushStatus{},
		ServiceAccounts:               map[host.Name]map[int][]string{},
	}
}

// JSON implements json.Marshaller, with a lock.
func (ps *PushContext) JSON() ([]byte, error) {
	if ps == nil {
		return []byte{'{', '}'}, nil
	}
	ps.proxyStatusMutex.RLock()
	defer ps.proxyStatusMutex.RUnlock()
	return json.MarshalIndent(ps, "", "    ")
}

// OnConfigChange is called when a config change is detected.
func (ps *PushContext) OnConfigChange() {
	LastPushMutex.Lock()
	LastPushStatus = ps
	LastPushMutex.Unlock()
	ps.UpdateMetrics()
}

// UpdateMetrics will update the prometheus metrics based on the
// current status of the push.
func (ps *PushContext) UpdateMetrics() {
	ps.proxyStatusMutex.RLock()
	defer ps.proxyStatusMutex.RUnlock()

	for _, pm := range metrics {
		mmap := ps.ProxyStatus[pm.Name()]
		pm.Record(float64(len(mmap)))
	}
}

// Services returns the list of services that are visible to a Proxy in a given config namespace
func (ps *PushContext) Services(proxy *Proxy) []*Service {
	// If proxy has a sidecar scope that is user supplied, then get the services from the sidecar scope
	// sidecarScope.config is nil if there is no sidecar scope for the namespace
	if proxy != nil && proxy.SidecarScope != nil && proxy.Type == SidecarProxy {
		return proxy.SidecarScope.Services()
	}

	out := make([]*Service, 0)

	// First add private services
	if proxy == nil {
		for _, privateServices := range ps.privateServicesByNamespace {
			out = append(out, privateServices...)
		}
	} else {
		out = append(out, ps.privateServicesByNamespace[proxy.ConfigNamespace]...)
	}

	// Second add public services
	out = append(out, ps.publicServices...)

	return out
}

// VirtualServices lists all virtual services bound to the specified gateways
// This replaces store.VirtualServices. Used only by the gateways
// Sidecars use the egressListener.VirtualServices().
func (ps *PushContext) VirtualServices(proxy *Proxy, gateways map[string]bool) []Config {
	configs := make([]Config, 0)
	out := make([]Config, 0)

	// filter out virtual services not reachable
	// First private virtual service
	if proxy == nil {
		for _, virtualSvcs := range ps.privateVirtualServicesByNamespace {
			configs = append(configs, virtualSvcs...)
		}
	} else {
		configs = append(configs, ps.privateVirtualServicesByNamespace[proxy.ConfigNamespace]...)
	}
	// Second public virtual service
	configs = append(configs, ps.publicVirtualServices...)

	for _, cfg := range configs {
		rule := cfg.Spec.(*networking.VirtualService)
		if len(rule.Gateways) == 0 {
			// This rule applies only to IstioMeshGateway
			if gateways[constants.IstioMeshGateway] {
				out = append(out, cfg)
			}
		} else {
			for _, g := range rule.Gateways {
				// note: Gateway names do _not_ use wildcard matching, so we do not use Name.Matches here
				if gateways[resolveGatewayName(g, cfg.ConfigMeta)] {
					out = append(out, cfg)
					break
				} else if g == constants.IstioMeshGateway && gateways[g] {
					// "mesh" gateway cannot be expanded into FQDN
					out = append(out, cfg)
					break
				}
			}
		}
	}

	return out
}

// getSidecarScope returns a SidecarScope object associated with the
// proxy. The SidecarScope object is a semi-processed view of the service
// registry, and config state associated with the sidecar crd. The scope contains
// a set of inbound and outbound listeners, services/configs per listener,
// etc. The sidecar scopes are precomputed in the initSidecarContext
// function based on the Sidecar API objects in each namespace. If there is
// no sidecar api object, a default sidecarscope is assigned to the
// namespace which enables connectivity to all services in the mesh.
//
// Callers can check if the sidecarScope is from user generated object or not
// by checking the sidecarScope.Config field, that contains the user provided config
func (ps *PushContext) getSidecarScope(proxy *Proxy, workloadLabels labels.Collection) *SidecarScope {

	// Find the most specific matching sidecar config from the proxy's
	// config namespace If none found, construct a sidecarConfig on the fly
	// that allows the sidecar to talk to any namespace (the default
	// behavior in the absence of sidecars).
	if sidecars, ok := ps.sidecarsByNamespace[proxy.ConfigNamespace]; ok {
		// TODO: logic to merge multiple sidecar resources
		// Currently we assume that there will be only one sidecar config for a namespace.
		var defaultSidecar *SidecarScope
		for _, wrapper := range sidecars {
			if wrapper.Config != nil {
				sidecar := wrapper.Config.Spec.(*networking.Sidecar)
				// if there is no workload selector, the config applies to all workloads
				// if there is a workload selector, check for matching workload labels
				if sidecar.GetWorkloadSelector() != nil {
					workloadSelector := labels.Instance(sidecar.GetWorkloadSelector().GetLabels())
					if !workloadLabels.IsSupersetOf(workloadSelector) {
						continue
					}
					return wrapper
				}
				defaultSidecar = wrapper
				continue
			}
			// Not sure when this can happen (Config = nil ?)
			if defaultSidecar != nil {
				return defaultSidecar // still return the valid one
			}
			return wrapper
		}
		if defaultSidecar != nil {
			return defaultSidecar // still return the valid one
		}
	}

	return DefaultSidecarScopeForNamespace(ps, proxy.ConfigNamespace)
}

// GetAllSidecarScopes returns a map of namespace and the set of SidecarScope
// object associated with the namespace. This will be used by the CDS code to
// precompute CDS output for each sidecar scope. Since we have a default sidecarscope
// for namespaces that dont explicitly have one, we are guaranteed to
// have the CDS output cached for every namespace/sidecar scope combo.
func (ps *PushContext) GetAllSidecarScopes() map[string][]*SidecarScope {
	return ps.sidecarsByNamespace
}

// DestinationRule returns a destination rule for a service name in a given domain.
func (ps *PushContext) DestinationRule(proxy *Proxy, service *Service) *Config {
	// FIXME: this code should be removed once the EDS issue is fixed
	if proxy == nil {
		if hostname, ok := MostSpecificHostMatch(service.Hostname, ps.allExportedDestRules.hosts); ok {
			return ps.allExportedDestRules.destRule[hostname].config
		}
		return nil
	}

	// If proxy has a sidecar scope that is user supplied, then get the destination rules from the sidecar scope
	// sidecarScope.config is nil if there is no sidecar scope for the namespace
	if proxy.SidecarScope != nil && proxy.Type == SidecarProxy {
		// If there is a sidecar scope for this proxy, return the destination rule
		// from the sidecar scope.
		return proxy.SidecarScope.DestinationRule(service.Hostname)
	}

	// If the proxy config namespace is same as the root config namespace
	// look for dest rules in the service's namespace first. This hack is needed
	// because sometimes, istio-system tends to become the root config namespace.
	// Destination rules are defined here for global purposes. We dont want these
	// catch all destination rules to be the only dest rule, when processing CDS for
	// proxies like the istio-ingressgateway or istio-egressgateway.
	// If there are no service specific dest rules, we will end up picking up the same
	// rules anyway, later in the code
	if proxy.ConfigNamespace != ps.Env.Mesh.RootNamespace {
		// search through the DestinationRules in proxy's namespace first
		if ps.namespaceLocalDestRules[proxy.ConfigNamespace] != nil {
			if hostname, ok := MostSpecificHostMatch(service.Hostname,
				ps.namespaceLocalDestRules[proxy.ConfigNamespace].hosts); ok {
				return ps.namespaceLocalDestRules[proxy.ConfigNamespace].destRule[hostname].config
			}
		}
	}

	svcNs := service.Attributes.Namespace

	// This can happen when finding the subset labels for a proxy in root namespace.
	// Because based on a pure cluster name, we do not know the service and
	// construct a fake service without setting Attributes at all.
	if svcNs == "" {
		for _, svc := range ps.Services(proxy) {
			if service.Hostname == svc.Hostname && svc.Attributes.Namespace != "" {
				svcNs = svc.Attributes.Namespace
				break
			}
		}
	}

	// if no private/public rule matched in the calling proxy's namespace,
	// check the target service's namespace for public rules
	if svcNs != "" && ps.namespaceExportedDestRules[svcNs] != nil {
		if hostname, ok := MostSpecificHostMatch(service.Hostname,
			ps.namespaceExportedDestRules[svcNs].hosts); ok {
			return ps.namespaceExportedDestRules[svcNs].destRule[hostname].config
		}
	}

	// if no public/private rule in calling proxy's namespace matched, and no public rule in the
	// target service's namespace matched, search for any public destination rule in the config root namespace
	// NOTE: This does mean that we are effectively ignoring private dest rules in the config root namespace
	if ps.namespaceExportedDestRules[ps.Env.Mesh.RootNamespace] != nil {
		if hostname, ok := MostSpecificHostMatch(service.Hostname,
			ps.namespaceExportedDestRules[ps.Env.Mesh.RootNamespace].hosts); ok {
			return ps.namespaceExportedDestRules[ps.Env.Mesh.RootNamespace].destRule[hostname].config
		}
	}

	return nil
}

// SubsetToLabels returns the labels associated with a subset of a given service.
func (ps *PushContext) SubsetToLabels(proxy *Proxy, subsetName string, hostname host.Name) labels.Collection {
	// empty subset
	if subsetName == "" {
		return nil
	}

	cfg := ps.DestinationRule(proxy, &Service{Hostname: hostname})
	if cfg == nil {
		return nil
	}

	rule := cfg.Spec.(*networking.DestinationRule)
	for _, subset := range rule.Subsets {
		if subset.Name == subsetName {
			return []labels.Instance{subset.Labels}
		}
	}

	return nil
}

// InitContext will initialize the data structures used for code generation.
// This should be called before starting the push, from the thread creating
// the push context.
func (ps *PushContext) InitContext(env *Environment) error {
	ps.Mutex.Lock()
	defer ps.Mutex.Unlock()
	if ps.initDone {
		return nil
	}
	ps.Env = env
	var err error

	// Must be initialized first
	// as initServiceRegistry/VirtualServices/Destrules
	// use the default export map
	ps.initDefaultExportMaps()

	if err = ps.initServiceRegistry(env); err != nil {
		return err
	}

	if err = ps.initVirtualServices(env); err != nil {
		return err
	}

	if err = ps.initDestinationRules(env); err != nil {
		return err
	}

	if err = ps.initAuthorizationPolicies(env); err != nil {
		rbacLog.Errorf("failed to initialize authorization policies: %v", err)
		return err
	}

	if err = ps.initEnvoyFilters(env); err != nil {
		return err
	}

	if err = ps.initGateways(env); err != nil {
		return err
	}

	// Must be initialized in the end
	if err = ps.initSidecarScopes(env); err != nil {
		return err
	}

	ps.initDone = true
	return nil
}

// Caches list of services in the registry, and creates a map
// of hostname to service
func (ps *PushContext) initServiceRegistry(env *Environment) error {
	services, err := env.Services()
	if err != nil {
		return err
	}
	// Sort the services in order of creation.
	allServices := sortServicesByCreationTime(services)
	for _, s := range allServices {
		ns := s.Attributes.Namespace
		if len(s.Attributes.ExportTo) == 0 {
			if ps.defaultServiceExportTo[visibility.Private] {
				ps.privateServicesByNamespace[ns] = append(ps.privateServicesByNamespace[ns], s)
			} else if ps.defaultServiceExportTo[visibility.Public] {
				ps.publicServices = append(ps.publicServices, s)
			}
		} else {
			if s.Attributes.ExportTo[visibility.Private] {
				ps.privateServicesByNamespace[ns] = append(ps.privateServicesByNamespace[ns], s)
			} else {
				ps.publicServices = append(ps.publicServices, s)
			}
		}
		if _, f := ps.ServiceByHostnameAndNamespace[s.Hostname]; !f {
			ps.ServiceByHostnameAndNamespace[s.Hostname] = map[string]*Service{}
		}
		ps.ServiceByHostnameAndNamespace[s.Hostname][s.Attributes.Namespace] = s
	}

	ps.initServiceAccounts(env, allServices)

	return nil
}

// sortServicesByCreationTime sorts the list of services in ascending order by their creation time (if available).
func sortServicesByCreationTime(services []*Service) []*Service {
	sort.SliceStable(services, func(i, j int) bool {
		return services[i].CreationTime.Before(services[j].CreationTime)
	})
	return services
}

// Caches list of service accounts in the registry
func (ps *PushContext) initServiceAccounts(env *Environment, services []*Service) {
	for _, svc := range services {
		if ps.ServiceAccounts[svc.Hostname] == nil {
			ps.ServiceAccounts[svc.Hostname] = map[int][]string{}
		}
		for _, port := range svc.Ports {
			if port.Protocol == protocol.UDP {
				continue
			}
			ps.ServiceAccounts[svc.Hostname][port.Port] = env.GetIstioServiceAccounts(svc, []int{port.Port})
		}
	}
}

// Caches list of virtual services
func (ps *PushContext) initVirtualServices(env *Environment) error {
	virtualServices, err := env.List(VirtualService.Type, NamespaceAll)
	if err != nil {
		return err
	}

	// values returned from ConfigStore.List are immutable.
	// Therefore, we make a copy
	vservices := make([]Config, len(virtualServices))

	for i := range vservices {
		vservices[i] = virtualServices[i].DeepCopy()
	}

	totalVirtualServices.Record(float64(len(virtualServices)))

	// TODO(rshriram): parse each virtual service and maintain a map of the
	// virtualservice name, the list of registry hosts in the VS and non
	// registry DNS names in the VS.  This should cut down processing in
	// the RDS code. See separateVSHostsAndServices in route/route.go
	sortConfigByCreationTime(vservices)

	// convert all shortnames in virtual services into FQDNs
	for _, r := range vservices {
		rule := r.Spec.(*networking.VirtualService)
		// resolve top level hosts
		for i, h := range rule.Hosts {
			rule.Hosts[i] = string(ResolveShortnameToFQDN(h, r.ConfigMeta))
		}
		// resolve gateways to bind to
		for i, g := range rule.Gateways {
			if g != constants.IstioMeshGateway {
				rule.Gateways[i] = resolveGatewayName(g, r.ConfigMeta)
			}
		}
		// resolve host in http route.destination, route.mirror
		for _, d := range rule.Http {
			for _, m := range d.Match {
				for i, g := range m.Gateways {
					if g != constants.IstioMeshGateway {
						m.Gateways[i] = resolveGatewayName(g, r.ConfigMeta)
					}
				}
			}
			for _, w := range d.Route {
				w.Destination.Host = string(ResolveShortnameToFQDN(w.Destination.Host, r.ConfigMeta))
			}
			if d.Mirror != nil {
				d.Mirror.Host = string(ResolveShortnameToFQDN(d.Mirror.Host, r.ConfigMeta))
			}
		}
		//resolve host in tcp route.destination
		for _, d := range rule.Tcp {
			for _, m := range d.Match {
				for i, g := range m.Gateways {
					if g != constants.IstioMeshGateway {
						m.Gateways[i] = resolveGatewayName(g, r.ConfigMeta)
					}
				}
			}
			for _, w := range d.Route {
				w.Destination.Host = string(ResolveShortnameToFQDN(w.Destination.Host, r.ConfigMeta))
			}
		}
		//resolve host in tls route.destination
		for _, tls := range rule.Tls {
			for _, m := range tls.Match {
				for i, g := range m.Gateways {
					if g != constants.IstioMeshGateway {
						m.Gateways[i] = resolveGatewayName(g, r.ConfigMeta)
					}
				}
			}
			for _, w := range tls.Route {
				w.Destination.Host = string(ResolveShortnameToFQDN(w.Destination.Host, r.ConfigMeta))
			}
		}
	}

	for _, virtualService := range vservices {
		ns := virtualService.Namespace
		rule := virtualService.Spec.(*networking.VirtualService)
		if len(rule.ExportTo) == 0 {
			// No exportTo in virtualService. Use the global default
			// TODO: We currently only honor ., * and ~
			if ps.defaultVirtualServiceExportTo[visibility.Private] {
				// add to local namespace only
				ps.privateVirtualServicesByNamespace[ns] = append(ps.privateVirtualServicesByNamespace[ns], virtualService)
			} else if ps.defaultVirtualServiceExportTo[visibility.Public] {
				ps.publicVirtualServices = append(ps.publicVirtualServices, virtualService)
			}
		} else {
			// TODO: we currently only process the first element in the array
			// and currently only consider . or * which maps to public/private
			if visibility.Instance(rule.ExportTo[0]) == visibility.Private {
				// add to local namespace only
				ps.privateVirtualServicesByNamespace[ns] = append(ps.privateVirtualServicesByNamespace[ns], virtualService)
			} else {
				// ~ is not valid in the exportTo fields in virtualServices, services, destination rules
				// and we currently only allow . or *. So treat this as public export
				ps.publicVirtualServices = append(ps.publicVirtualServices, virtualService)
			}
		}
	}

	return nil
}

func (ps *PushContext) initDefaultExportMaps() {
	ps.defaultDestinationRuleExportTo = make(map[visibility.Instance]bool)
	if ps.Env.Mesh.DefaultDestinationRuleExportTo != nil {
		for _, e := range ps.Env.Mesh.DefaultDestinationRuleExportTo {
			ps.defaultDestinationRuleExportTo[visibility.Instance(e)] = true
		}
	} else {
		// default to *
		ps.defaultDestinationRuleExportTo[visibility.Public] = true
	}

	ps.defaultServiceExportTo = make(map[visibility.Instance]bool)
	if ps.Env.Mesh.DefaultServiceExportTo != nil {
		for _, e := range ps.Env.Mesh.DefaultServiceExportTo {
			ps.defaultServiceExportTo[visibility.Instance(e)] = true
		}
	} else {
		ps.defaultServiceExportTo[visibility.Public] = true
	}

	ps.defaultVirtualServiceExportTo = make(map[visibility.Instance]bool)
	if ps.Env.Mesh.DefaultVirtualServiceExportTo != nil {
		for _, e := range ps.Env.Mesh.DefaultVirtualServiceExportTo {
			ps.defaultVirtualServiceExportTo[visibility.Instance(e)] = true
		}
	} else {
		ps.defaultVirtualServiceExportTo[visibility.Public] = true
	}
}

// initSidecarScopes synthesizes Sidecar CRDs into objects called
// SidecarScope.  The SidecarScope object is a semi-processed view of the
// service registry, and config state associated with the sidecar CRD. The
// scope contains a set of inbound and outbound listeners, services/configs
// per listener, etc. The sidecar scopes are precomputed based on the
// Sidecar API objects in each namespace. If there is no sidecar api object
// for a namespace, a default sidecarscope is assigned to the namespace
// which enables connectivity to all services in the mesh.
//
// When proxies connect to Pilot, we identify the sidecar scope associated
// with the proxy and derive listeners/routes/clusters based on the sidecar
// scope.
func (ps *PushContext) initSidecarScopes(env *Environment) error {
	sidecarConfigs, err := env.List(Sidecar.Type, NamespaceAll)
	if err != nil {
		return err
	}

	sortConfigByCreationTime(sidecarConfigs)

	sidecarConfigWithSelector := make([]Config, 0)
	sidecarConfigWithoutSelector := make([]Config, 0)
	for _, sidecarConfig := range sidecarConfigs {
		sidecar := sidecarConfig.Spec.(*networking.Sidecar)
		if sidecar.WorkloadSelector != nil {
			sidecarConfigWithSelector = append(sidecarConfigWithSelector, sidecarConfig)
		} else {
			sidecarConfigWithoutSelector = append(sidecarConfigWithoutSelector, sidecarConfig)
		}
	}

	sidecarNum := len(sidecarConfigs)
	sidecarConfigs = make([]Config, 0, sidecarNum)
	sidecarConfigs = append(sidecarConfigs, sidecarConfigWithSelector...)
	sidecarConfigs = append(sidecarConfigs, sidecarConfigWithoutSelector...)

	ps.sidecarsByNamespace = make(map[string][]*SidecarScope, sidecarNum)
	for _, sidecarConfig := range sidecarConfigs {
		sidecarConfig := sidecarConfig
		ps.sidecarsByNamespace[sidecarConfig.Namespace] = append(ps.sidecarsByNamespace[sidecarConfig.Namespace],
			ConvertToSidecarScope(ps, &sidecarConfig, sidecarConfig.Namespace))
	}

	// Hold reference root namespace's sidecar config
	// Root namespace can have only one sidecar config object
	// Currently we expect that it has no workloadSelectors
	var rootNSConfig *Config
	if env.Mesh.RootNamespace != "" {
		for _, sidecarConfig := range sidecarConfigs {
			if sidecarConfig.Namespace == env.Mesh.RootNamespace &&
				sidecarConfig.Spec.(*networking.Sidecar).WorkloadSelector == nil {
				rootNSConfig = &sidecarConfig
				break
			}
		}
	}

	// build sidecar scopes for other namespaces that dont have a sidecar CRD object.
	// Derive the sidecar scope from the root namespace's sidecar object if present. Else fallback
	// to the default Istio behavior mimicked by the DefaultSidecarScopeForNamespace function.
	for _, nsMap := range ps.ServiceByHostnameAndNamespace {
		for ns := range nsMap {
			if len(ps.sidecarsByNamespace[ns]) == 0 {
				// use the contents from the root namespace or the default if there is no root namespace
				ps.sidecarsByNamespace[ns] = []*SidecarScope{ConvertToSidecarScope(ps, rootNSConfig, ns)}
			}
		}
	}

	return nil
}

// Split out of DestinationRule expensive conversions - once per push.
func (ps *PushContext) initDestinationRules(env *Environment) error {
	configs, err := env.List(DestinationRule.Type, NamespaceAll)
	if err != nil {
		return err
	}
	ps.SetDestinationRules(configs)
	return nil
}

// SetDestinationRules is updates internal structures using a set of configs.
// Split out of DestinationRule expensive conversions, computed once per push.
// This also allows tests to inject a config without having the mock.
// This will not work properly for Sidecars, which will precompute their destination rules on init
func (ps *PushContext) SetDestinationRules(configs []Config) {
	// Sort by time first. So if two destination rule have top level traffic policies
	// we take the first one.
	sortConfigByCreationTime(configs)
	namespaceLocalDestRules := make(map[string]*processedDestRules)
	namespaceExportedDestRules := make(map[string]*processedDestRules)
	allExportedDestRules := &processedDestRules{
		hosts:    make([]host.Name, 0),
		destRule: map[host.Name]*combinedDestinationRule{},
	}

	for i := range configs {
		rule := configs[i].Spec.(*networking.DestinationRule)
		rule.Host = string(ResolveShortnameToFQDN(rule.Host, configs[i].ConfigMeta))
		// Store in an index for the config's namespace
		// a proxy from this namespace will first look here for the destination rule for a given service
		// This pool consists of both public/private destination rules.
		// TODO: when exportTo is fully supported, only add the rule here if exportTo is '.'
		// The global exportTo doesn't matter here (its either . or * - both of which are applicable here)
		if _, exist := namespaceLocalDestRules[configs[i].Namespace]; !exist {
			namespaceLocalDestRules[configs[i].Namespace] = &processedDestRules{
				hosts:    make([]host.Name, 0),
				destRule: map[host.Name]*combinedDestinationRule{},
			}
		}
		// Merge this destination rule with any public/private dest rules for same host in the same namespace
		// If there are no duplicates, the dest rule will be added to the list
		namespaceLocalDestRules[configs[i].Namespace].hosts = ps.combineSingleDestinationRule(
			namespaceLocalDestRules[configs[i].Namespace].hosts,
			namespaceLocalDestRules[configs[i].Namespace].destRule,
			configs[i])

		isPubliclyExported := false
		if len(rule.ExportTo) == 0 {
			// No exportTo in destinationRule. Use the global default
			// TODO: We currently only honor ., * and ~
			if ps.defaultDestinationRuleExportTo[visibility.Public] {
				isPubliclyExported = true
			}
		} else {
			// TODO: we currently only process the first element in the array
			// and currently only consider . or * which maps to public/private
			if visibility.Instance(rule.ExportTo[0]) != visibility.Private {
				// ~ is not valid in the exportTo fields in virtualServices, services, destination rules
				// and we currently only allow . or *. So treat this as public export
				isPubliclyExported = true
			}
		}

		if isPubliclyExported {
			if _, exist := namespaceExportedDestRules[configs[i].Namespace]; !exist {
				namespaceExportedDestRules[configs[i].Namespace] = &processedDestRules{
					hosts:    make([]host.Name, 0),
					destRule: map[host.Name]*combinedDestinationRule{},
				}
			}
			// Merge this destination rule with any public dest rule for the same host in the same namespace
			// If there are no duplicates, the dest rule will be added to the list
			namespaceExportedDestRules[configs[i].Namespace].hosts = ps.combineSingleDestinationRule(
				namespaceExportedDestRules[configs[i].Namespace].hosts,
				namespaceExportedDestRules[configs[i].Namespace].destRule,
				configs[i])

			// Merge this destination rule with any public dest rule for the same host
			// across all namespaces. If there are no duplicates, the dest rule will be added to the list
			allExportedDestRules.hosts = ps.combineSingleDestinationRule(
				allExportedDestRules.hosts, allExportedDestRules.destRule, configs[i])
		}
	}

	// presort it so that we don't sort it for each DestinationRule call.
	// sort.Sort for Hostnames will automatically sort from the most specific to least specific
	for ns := range namespaceLocalDestRules {
		sort.Sort(host.Names(namespaceLocalDestRules[ns].hosts))
	}
	for ns := range namespaceExportedDestRules {
		sort.Sort(host.Names(namespaceExportedDestRules[ns].hosts))
	}
	sort.Sort(host.Names(allExportedDestRules.hosts))

	ps.namespaceLocalDestRules = namespaceLocalDestRules
	ps.namespaceExportedDestRules = namespaceExportedDestRules
	ps.allExportedDestRules = allExportedDestRules
}

func (ps *PushContext) initAuthorizationPolicies(env *Environment) error {
	var err error
	if ps.AuthzPolicies, err = NewAuthzPolicies(env); err != nil {
		rbacLog.Errorf("failed to initialize authorization policies: %v", err)
		return err
	}
	return nil
}

// pre computes envoy filters per namespace
func (ps *PushContext) initEnvoyFilters(env *Environment) error {
	envoyFilterConfigs, err := env.List(EnvoyFilter.Type, NamespaceAll)
	if err != nil {
		return err
	}

	sortConfigByCreationTime(envoyFilterConfigs)

	ps.envoyFiltersByNamespace = make(map[string][]*EnvoyFilterWrapper)
	for _, envoyFilterConfig := range envoyFilterConfigs {
		efw := convertToEnvoyFilterWrapper(&envoyFilterConfig)
		if _, exists := ps.envoyFiltersByNamespace[envoyFilterConfig.Namespace]; !exists {
			ps.envoyFiltersByNamespace[envoyFilterConfig.Namespace] = make([]*EnvoyFilterWrapper, 0)
		}
		ps.envoyFiltersByNamespace[envoyFilterConfig.Namespace] = append(ps.envoyFiltersByNamespace[envoyFilterConfig.Namespace], efw)
	}
	return nil
}

func (ps *PushContext) EnvoyFilters(proxy *Proxy) []*EnvoyFilterWrapper {
	// this should never happen
	if proxy == nil {
		return nil
	}
	out := make([]*EnvoyFilterWrapper, 0)
	// EnvoyFilters supports inheritance (global ones plus namespace local ones).
	// First get all the filter configs from the config root namespace
	// and then add the ones from proxy's own namespace
	if ps.Env.Mesh.RootNamespace != "" {
		// if there is no workload selector, the config applies to all workloads
		// if there is a workload selector, check for matching workload labels
		for _, efw := range ps.envoyFiltersByNamespace[ps.Env.Mesh.RootNamespace] {
			if efw.workloadSelector == nil || proxy.WorkloadLabels.IsSupersetOf(efw.workloadSelector) {
				out = append(out, efw)
			}
		}
	}

	// To prevent duplicate envoyfilters in case root namespace equals proxy's namespace
	if proxy.ConfigNamespace != ps.Env.Mesh.RootNamespace {
		for _, efw := range ps.envoyFiltersByNamespace[proxy.ConfigNamespace] {
			if efw.workloadSelector == nil || proxy.WorkloadLabels.IsSupersetOf(efw.workloadSelector) {
				out = append(out, efw)
			}
		}
	}

	return out
}

// pre computes gateways per namespace
func (ps *PushContext) initGateways(env *Environment) error {
	gatewayConfigs, err := env.List(Gateway.Type, NamespaceAll)
	if err != nil {
		return err
	}

	sortConfigByCreationTime(gatewayConfigs)

	ps.allGateways = gatewayConfigs
	ps.gatewaysByNamespace = make(map[string][]Config)
	for _, gatewayConfig := range gatewayConfigs {
		if _, exists := ps.gatewaysByNamespace[gatewayConfig.Namespace]; !exists {
			ps.gatewaysByNamespace[gatewayConfig.Namespace] = make([]Config, 0)
		}
		ps.gatewaysByNamespace[gatewayConfig.Namespace] = append(ps.gatewaysByNamespace[gatewayConfig.Namespace], gatewayConfig)
	}
	return nil
}

func (ps *PushContext) mergeGateways(proxy *Proxy) *MergedGateway {
	// this should never happen
	if proxy == nil {
		return nil
	}
	out := make([]Config, 0)

	var configs []Config
	if features.ScopeGatewayToNamespace.Get() {
		configs = ps.gatewaysByNamespace[proxy.ConfigNamespace]
	} else {
		configs = ps.allGateways
	}

	for _, cfg := range configs {
		gw := cfg.Spec.(*networking.Gateway)
		if gw.GetSelector() == nil {
			// no selector. Applies to all workloads asking for the gateway
			out = append(out, cfg)
		} else {
			gatewaySelector := labels.Instance(gw.GetSelector())
			if proxy.WorkloadLabels.IsSupersetOf(gatewaySelector) {
				out = append(out, cfg)
			}
		}
	}

	if len(out) == 0 {
		return nil
	}
	return MergeGateways(out...)
}
