// connectProxyConfigWatch is used to watch for changes to Connect managed proxy
// configuration. Note that this state is agent-local so the watch mechanism
// uses `hash` rather than `index` for deciding whether to block.
178941
func connectProxyConfigWatch(params map[string]interface{}) (WatcherFunc, error) {
	// We don't support consistency modes since it's agent local data

	var proxyServiceID string
	if err := assignValue(params, "proxy_service_id", &proxyServiceID); err != nil {
		return nil, err
	}

	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		agent := p.client.Agent()
		opts := makeQueryOptionsWithContext(p, false)
		defer p.cancelFunc()

		config, _, err := agent.ConnectProxyConfig(proxyServiceID, &opts)
		if err != nil {
			return nil, nil, err
		}

		// Return string ContentHash since we don't have Raft indexes to block on.
		return WaitHashVal(config.ContentHash), config, err
	}
	return fn, nil
}
505065
func (p *ExamplePlugin) subscribeWatcher() (err error) {
	prefix := etcdKeyPrefix(p.ServiceLabel.GetAgentLabel())
	p.Log.Infof("Prefix: %v", prefix)

	p.watchDataReg, err = p.Watcher.Watch("ExamplePlugin",
		p.changeChannel, p.resyncChannel, prefix)
	if err != nil {
		return err
	}

	p.Log.Info("KeyValProtoWatcher subscribed")

	return nil
}
229150
func NewProxyWatcher(ch chan Event) *ProxyWatcher {
	return &ProxyWatcher{
		result:  ch,
		stopCh:  make(chan struct{}),
		stopped: false,
	}
}
527044
func (sdn *OpenShiftSDN) runProxy() {
	protocol := utiliptables.ProtocolIpv4
	bindAddr := net.ParseIP(sdn.ProxyConfig.BindAddress)
	if bindAddr.To4() == nil {
		protocol = utiliptables.ProtocolIpv6
	}

	portRange := utilnet.ParsePortRangeOrDie(sdn.ProxyConfig.PortRange)

	hostname, err := utilnode.GetHostname(sdn.ProxyConfig.HostnameOverride)
	if err != nil {
		klog.Fatalf("Unable to get hostname: %v", err)
	}

	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartRecordingToSink(&kv1core.EventSinkImpl{Interface: sdn.informers.KubeClient.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "kube-proxy", Host: hostname})

	execer := utilexec.New()
	dbus := utildbus.New()
	iptInterface := utiliptables.New(execer, dbus, protocol)

	var proxier proxy.ProxyProvider
	var servicesHandler pconfig.ServiceHandler
	var endpointsHandler pconfig.EndpointsHandler
	var healthzServer *healthcheck.HealthzServer
	if len(sdn.ProxyConfig.HealthzBindAddress) > 0 {
		nodeRef := &v1.ObjectReference{
			Kind:      "Node",
			Name:      hostname,
			UID:       types.UID(hostname),
			Namespace: "",
		}
		healthzServer = healthcheck.NewDefaultHealthzServer(sdn.ProxyConfig.HealthzBindAddress, 2*sdn.ProxyConfig.IPTables.SyncPeriod.Duration, recorder, nodeRef)
	}

	switch sdn.ProxyConfig.Mode {
	case kubeproxyconfig.ProxyModeIPTables:
		klog.V(0).Info("Using iptables Proxier.")
		if bindAddr.Equal(net.IPv4zero) {
			var err error
			bindAddr, err = getNodeIP(sdn.informers.KubeClient.CoreV1(), hostname)
			if err != nil {
				klog.Fatalf("Unable to get a bind address: %v", err)
			}
		}
		if sdn.ProxyConfig.IPTables.MasqueradeBit == nil {
			// IPTablesMasqueradeBit must be specified or defaulted.
			klog.Fatalf("Unable to read IPTablesMasqueradeBit from config")
		}
		proxierIptables, err := iptables.NewProxier(
			iptInterface,
			utilsysctl.New(),
			execer,
			sdn.ProxyConfig.IPTables.SyncPeriod.Duration,
			sdn.ProxyConfig.IPTables.MinSyncPeriod.Duration,
			sdn.ProxyConfig.IPTables.MasqueradeAll,
			int(*sdn.ProxyConfig.IPTables.MasqueradeBit),
			sdn.ProxyConfig.ClusterCIDR,
			hostname,
			bindAddr,
			recorder,
			healthzServer,
			sdn.ProxyConfig.NodePortAddresses,
		)
		metrics.RegisterMetrics()

		if err != nil {
			klog.Fatalf("error: Could not initialize Kubernetes Proxy. You must run this process as root (and if containerized, in the host network namespace as privileged) to use the service proxy: %v", err)
		}
		proxier = proxierIptables
		endpointsHandler = proxierIptables
		servicesHandler = proxierIptables
		// No turning back. Remove artifacts that might still exist from the userspace Proxier.
		klog.V(0).Info("Tearing down userspace rules.")
		userspace.CleanupLeftovers(iptInterface)
	case kubeproxyconfig.ProxyModeUserspace:
		klog.V(0).Info("Using userspace Proxier.")
		// This is a proxy.LoadBalancer which NewProxier needs but has methods we don't need for
		// our config.EndpointsHandler.
		loadBalancer := userspace.NewLoadBalancerRR()
		// set EndpointsHandler to our loadBalancer
		endpointsHandler = loadBalancer

		execer := utilexec.New()
		proxierUserspace, err := userspace.NewProxier(
			loadBalancer,
			bindAddr,
			iptInterface,
			execer,
			*portRange,
			sdn.ProxyConfig.IPTables.SyncPeriod.Duration,
			sdn.ProxyConfig.IPTables.MinSyncPeriod.Duration,
			sdn.ProxyConfig.UDPIdleTimeout.Duration,
			sdn.ProxyConfig.NodePortAddresses,
		)
		if err != nil {
			klog.Fatalf("error: Could not initialize Kubernetes Proxy. You must run this process as root (and if containerized, in the host network namespace as privileged) to use the service proxy: %v", err)
		}
		proxier = proxierUserspace
		servicesHandler = proxierUserspace
		// Remove artifacts from the pure-iptables Proxier.
		klog.V(0).Info("Tearing down pure-iptables proxy rules.")
		iptables.CleanupLeftovers(iptInterface)
	default:
		klog.Fatalf("Unknown proxy mode %q", sdn.ProxyConfig.Mode)
	}

	// Create configs (i.e. Watches for Services and Endpoints)
	// Note: RegisterHandler() calls need to happen before creation of Sources because sources
	// only notify on changes, and the initial update (on process start) may be lost if no handlers
	// are registered yet.
	serviceConfig := pconfig.NewServiceConfig(
		sdn.informers.KubeInformers.Core().V1().Services(),
		sdn.ProxyConfig.ConfigSyncPeriod.Duration,
	)

	if sdn.NodeConfig.EnableUnidling {
		unidlingLoadBalancer := userspace.NewLoadBalancerRR()
		signaler := unidler.NewEventSignaler(recorder)
		unidlingUserspaceProxy, err := unidler.NewUnidlerProxier(unidlingLoadBalancer, bindAddr, iptInterface, execer, *portRange, sdn.ProxyConfig.IPTables.SyncPeriod.Duration, sdn.ProxyConfig.IPTables.MinSyncPeriod.Duration, sdn.ProxyConfig.UDPIdleTimeout.Duration, sdn.ProxyConfig.NodePortAddresses, signaler)
		if err != nil {
			klog.Fatalf("error: Could not initialize Kubernetes Proxy. You must run this process as root (and if containerized, in the host network namespace as privileged) to use the service proxy: %v", err)
		}
		hybridProxier, err := hybrid.NewHybridProxier(
			unidlingLoadBalancer,
			unidlingUserspaceProxy,
			endpointsHandler,
			servicesHandler,
			proxier,
			unidlingUserspaceProxy,
			sdn.ProxyConfig.IPTables.SyncPeriod.Duration,
			sdn.informers.KubeInformers.Core().V1().Services().Lister(),
		)
		if err != nil {
			klog.Fatalf("error: Could not initialize Kubernetes Proxy. You must run this process as root (and if containerized, in the host network namespace as privileged) to use the service proxy: %v", err)
		}
		endpointsHandler = hybridProxier
		servicesHandler = hybridProxier
		proxier = hybridProxier
	}

	iptInterface.AddReloadFunc(proxier.Sync)
	serviceConfig.RegisterEventHandler(servicesHandler)
	go serviceConfig.Run(utilwait.NeverStop)

	endpointsConfig := pconfig.NewEndpointsConfig(
		sdn.informers.KubeInformers.Core().V1().Endpoints(),
		sdn.ProxyConfig.ConfigSyncPeriod.Duration,
	)
	// customized handling registration that inserts a filter if needed
	if err := sdn.OsdnProxy.Start(endpointsHandler); err != nil {
		klog.Fatalf("error: node proxy plugin startup failed: %v", err)
	}
	endpointsHandler = sdn.OsdnProxy
	endpointsConfig.RegisterEventHandler(endpointsHandler)
	go endpointsConfig.Run(utilwait.NeverStop)

	// Start up healthz server
	if len(sdn.ProxyConfig.HealthzBindAddress) > 0 {
		healthzServer.Run()
	}

	// Start up a metrics server if requested
	if len(sdn.ProxyConfig.MetricsBindAddress) > 0 {
		mux := http.NewServeMux()
		mux.HandleFunc("/proxyMode", func(w http.ResponseWriter, r *http.Request) {
			fmt.Fprintf(w, "%s", sdn.ProxyConfig.Mode)
		})
		mux.Handle("/metrics", prometheus.Handler())
		go utilwait.Until(func() {
			err := http.ListenAndServe(sdn.ProxyConfig.MetricsBindAddress, mux)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("starting metrics server failed: %v", err))
			}
		}, 5*time.Second, utilwait.NeverStop)
	}

	// periodically sync k8s iptables rules
	go utilwait.Forever(proxier.SyncLoop, 0)
	klog.Infof("Started Kubernetes Proxy on %s", sdn.ProxyConfig.BindAddress)
}
417825
func (c *FakeIngresses) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(ingressesResource, c.ns, opts))

}
1281997
func (adapter *Adapter) Watch(resyncName string, changeChan chan datasync.ChangeEvent,
	resyncChan chan datasync.ResyncEvent, keyPrefixes ...string) (datasync.WatchRegistration, error) {

	if adapter.Watcher != nil {
		return adapter.Watcher.Watch(resyncName, changeChan, resyncChan, keyPrefixes...)
	}
	logrus.DefaultLogger().Debug("KeyValProtoWatcher is nil")

	return nil, nil
}
1320204
func newProxy(addr string, tlsConfig *tls.Config) (*Proxy, error) {
	p := &Proxy{
		addr: addr,
	}

	if tlsConfig != nil {
		p.dialOpts = append(p.dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
	} else {
		p.dialOpts = append(p.dialOpts, grpc.WithInsecure())
	}

	conn, err := grpc.Dial(p.addr, p.dialOpts...)
	if err != nil {
		return nil, err
	}
	p.client = pb.NewDnsServiceClient(conn)

	return p, nil
}
22754
func (c *Client) WatchPrefix(prefix string, keys []string, waitIndex uint64, stopChan chan bool) (uint64, error) {
	<-stopChan
	return 0, nil
}
859127
func (p *AuthProxy) Listen() error {
	mux := bone.New()
	proxy := &httputil.ReverseProxy{
		Transport: p.t,
		Director: func(r *http.Request) {
			r.URL.Scheme = p.u.Scheme
			r.URL.Host = p.u.Host
			r.Host = p.u.Host
			r.URL.Path = r.URL.Path[6:]

			if p.sess.HasToken() {
				tok := string(p.sess.Token())
				r.Header["Authorization"] = []string{"Bearer " + tok}
			}

			r.Header["User-Agent"] = []string{"Torus-Daemon/" + p.c.Version}
			r.Header["X-Registry-Version"] = []string{p.c.APIVersion}
		},
	}

	go p.o.Start()

	mux.HandleFunc("/proxy/", proxyCanceler(proxy))
	mux.SubRoute("/v1", routes.NewRouteMux(p.c, p.sess, p.db, p.t, p.o, p.client, p.logic, p.updates))

	h := httpdown.HTTP{}
	p.s = h.Serve(&http.Server{Handler: requestIDHandler(loggingHandler(mux))}, p.l)

	return p.s.Wait()
}
373830
func NewProxy(addr, trans string) *Proxy {
	p := &Proxy{
		addr:      addr,
		fails:     0,
		probe:     up.New(),
		transport: newTransport(addr),
	}
	p.health = NewHealthChecker(trans)
	runtime.SetFinalizer(p, (*Proxy).finalizer)
	return p
}
822964
func (client *Client) CreateProxy(name, listen, upstream string) (*Proxy, error) {
	proxy := &Proxy{
		Name:     name,
		Listen:   listen,
		Upstream: upstream,
		Enabled:  true,
		client:   client,
	}

	err := proxy.Save()
	if err != nil {
		return nil, err
	}

	return proxy, nil
}
810136
func (w *proxyWorker) SetUp() (watcher.NotifyWatcher, error) {
	// We need to set this up initially as the NotifyWorker sucks up the first
	// event.
	err := w.onChange()
	if err != nil {
		return nil, err
	}
	w.first = false
	return w.config.API.WatchForProxyConfigAndAPIHostPortChanges()
}
1023573
func defaultProxyCommand(agentCfg *config.RuntimeConfig) ([]string, error) {
	// Get the path to the current executable. This is cached once by the
	// library so this is effectively just a variable read.
	execPath, err := os.Executable()
	if err != nil {
		return nil, err
	}

	// "consul connect proxy" default value for managed daemon proxy
	cmd := []string{execPath, "connect", "proxy"}

	if agentCfg != nil && agentCfg.LogLevel != "INFO" {
		cmd = append(cmd, "-log-level", agentCfg.LogLevel)
	}
	return cmd, nil
}
740031
func (c *FakeBrokerTemplateInstances) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(brokertemplateinstancesResource, opts))
}
439383
func keyPrefixWatch(params map[string]interface{}) (WatcherFunc, error) {
	stale := false
	if err := assignValueBool(params, "stale", &stale); err != nil {
		return nil, err
	}

	var prefix string
	if err := assignValue(params, "prefix", &prefix); err != nil {
		return nil, err
	}
	if prefix == "" {
		return nil, fmt.Errorf("Must specify a single prefix to watch")
	}
	fn := func(p *Plan) (BlockingParamVal, interface{}, error) {
		kv := p.client.KV()
		opts := makeQueryOptionsWithContext(p, stale)
		defer p.cancelFunc()
		pairs, meta, err := kv.List(prefix, &opts)
		if err != nil {
			return nil, nil, err
		}
		return WaitIndexVal(meta.LastIndex), pairs, err
	}
	return fn, nil
}
356433
func (c *FakeSecurityContextConstraints) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewRootWatchAction(securitycontextconstraintsResource, opts))
}
1105324
func (c *Config) SnapProxySettings() proxy.Settings {
	return proxy.Settings{
		Http:  c.SnapHTTPProxy(),
		Https: c.SnapHTTPSProxy(),
	}
}