// OnUpdate manages the active set of service proxies.
// Active service proxies are reinitialized if found in the update set or
// shutdown if missing from the update set.
func (proxier *Proxier) OnUpdate(services []api.Service) {
	glog.V(4).Infof("Received update notice: %+v", services)
	activeServices := make(map[types.NamespacedName]bool) // use a map as a set
	for _, service := range services {
		serviceName := types.NamespacedName{service.Namespace, service.Name}
		activeServices[serviceName] = true
		info, exists := proxier.getServiceInfo(serviceName)
		serviceIP := net.ParseIP(service.Spec.PortalIP)
		// TODO: check health of the socket?  What if ProxyLoop exited?
		if exists && info.portalPort == service.Spec.Port && info.portalIP.Equal(serviceIP) {
			continue
		}
		if exists && (info.portalPort != service.Spec.Port || !info.portalIP.Equal(serviceIP) || !ipsEqual(service.Spec.PublicIPs, info.publicIP)) {
			glog.V(4).Infof("Something changed for service %q: stopping it", serviceName.String())
			err := proxier.closePortal(serviceName, info)
			if err != nil {
				glog.Errorf("Failed to close portal for %q: %v", serviceName, err)
			}
			err = proxier.stopProxy(serviceName, info)
			if err != nil {
				glog.Errorf("Failed to stop service %q: %v", serviceName, err)
			}
		}
		glog.V(1).Infof("Adding new service %q at %s:%d/%s", serviceName, serviceIP, service.Spec.Port, service.Spec.Protocol)
		info, err := proxier.addServiceOnPort(serviceName, service.Spec.Protocol, 0, udpIdleTimeout)
		if err != nil {
			glog.Errorf("Failed to start proxy for %q: %v", serviceName, err)
			continue
		}
		info.portalIP = serviceIP
		info.portalPort = service.Spec.Port
		info.publicIP = service.Spec.PublicIPs
		info.sessionAffinityType = service.Spec.SessionAffinity
		// TODO: paramaterize this in the types api file as an attribute of sticky session.   For now it's hardcoded to 3 hours.
		info.stickyMaxAgeMinutes = 180
		glog.V(4).Infof("info: %+v", info)

		err = proxier.openPortal(serviceName, info)
		if err != nil {
			glog.Errorf("Failed to open portal for %q: %v", serviceName, err)
		}
		proxier.loadBalancer.NewService(serviceName, info.sessionAffinityType, info.stickyMaxAgeMinutes)
	}
	proxier.mu.Lock()
	defer proxier.mu.Unlock()
	for name, info := range proxier.serviceMap {
		if !activeServices[name] {
			glog.V(1).Infof("Stopping service %q", name)
			err := proxier.closePortal(name, info)
			if err != nil {
				glog.Errorf("Failed to close portal for %q: %v", name, err)
			}
			err = proxier.stopProxyInternal(name, info)
			if err != nil {
				glog.Errorf("Failed to stop service %q: %v", name, err)
			}
		}
	}
}

// Returns an error if processing the delta failed, along with a boolean
// indicator of whether the processing should be retried.
func (s *ServiceController) processDelta(delta *cache.Delta) (error, bool) {
	service, ok := delta.Object.(*api.Service)
	var namespacedName types.NamespacedName
	var cachedService *cachedService
	if !ok {
		// If the DeltaFIFO saw a key in our cache that it didn't know about, it
		// can send a deletion with an unknown state. Grab the service from our
		// cache for deleting.
		key, ok := delta.Object.(cache.DeletedFinalStateUnknown)
		if !ok {
			return fmt.Errorf("Delta contained object that wasn't a service or a deleted key: %+v", delta), notRetryable
		}
		cachedService, ok = s.cache.get(key.Key)
		if !ok {
			return fmt.Errorf("Service %s not in cache even though the watcher thought it was. Ignoring the deletion.", key), notRetryable
		}
		service = cachedService.service
		delta.Object = cachedService.service
		namespacedName = types.NamespacedName{service.Namespace, service.Name}
	} else {
		namespacedName.Namespace = service.Namespace
		namespacedName.Name = service.Name
		cachedService = s.cache.getOrCreate(namespacedName.String())
	}
	glog.V(2).Infof("Got new %s delta for service: %+v", delta.Type, service)

	// Ensure that no other goroutine will interfere with our processing of the
	// service.
	cachedService.mu.Lock()
	defer cachedService.mu.Unlock()

	// TODO: Handle added, updated, and sync differently?
	switch delta.Type {
	case cache.Added:
		fallthrough
	case cache.Updated:
		fallthrough
	case cache.Sync:
		err, retry := s.createLoadBalancerIfNeeded(namespacedName, service, cachedService.service)
		if err != nil {
			return err, retry
		}
		// Always update the cache upon success.
		// NOTE: Since we update the cached service if and only if we successully
		// processed it, a cached service being nil implies that it hasn't yet
		// been successfully processed.
		cachedService.service = service
		s.cache.set(namespacedName.String(), cachedService)
	case cache.Deleted:
		err := s.balancer.EnsureTCPLoadBalancerDeleted(s.loadBalancerName(service), s.zone.Region)
		if err != nil {
			return err, retryable
		}
		s.cache.delete(namespacedName.String())
	default:
		glog.Errorf("Unexpected delta type: %v", delta.Type)
	}
	return nil, notRetryable
}

// Build a slice of iptables args that are common to from-container and from-host portal rules.
func iptablesCommonPortalArgs(destIP net.IP, destPort int, protocol api.Protocol, service types.NamespacedName) []string {
	// This list needs to include all fields as they are eventually spit out
	// by iptables-save.  This is because some systems do not support the
	// 'iptables -C' arg, and so fall back on parsing iptables-save output.
	// If this does not match, it will not pass the check.  For example:
	// adding the /32 on the destination IP arg is not strictly required,
	// but causes this list to not match the final iptables-save output.
	// This is fragile and I hope one day we can stop supporting such old
	// iptables versions.
	args := []string{
		"-m", "comment",
		"--comment", service.String(),
		"-p", strings.ToLower(string(protocol)),
		"-m", strings.ToLower(string(protocol)),
		"-d", fmt.Sprintf("%s/32", destIP.String()),
		"--dport", fmt.Sprintf("%d", destPort),
	}
	return args
}