// clustersHandler returns cluster information for all Sidecar services. It
// implements the Envoy CDS API V1.
519149
func (s *EnvoyApi) clustersHandler(response http.ResponseWriter, req *http.Request, params map[string]string) {
	defer req.Body.Close()

	response.Header().Set("Content-Type", "application/json")

	clusters := s.EnvoyClustersFromState()

	log.Debugf("Reporting Envoy cluster information for cluster '%s' and node '%s'",
		params["service_cluster"], params["service_node"])

	result := CDSResult{clusters}

	jsonBytes, err := result.MarshalJSON()
	defer ffjson.Pool(jsonBytes)
	if err != nil {
		log.Errorf("Error marshaling state in servicesHandler: %s", err.Error())
		sendJsonError(response, 500, "Internal server error")
		return
	}

	response.Write(jsonBytes)
}
1056990
func NewClusterStateService(client *Client) *ClusterStateService {
	return &ClusterStateService{
		client:  client,
		indices: make([]string, 0),
		metrics: make([]string, 0),
	}
}
1180712
func clusterDialer(remoteCluster reversetunnel.RemoteSite) auth.DialContext {
	return func(in context.Context, network, _ string) (net.Conn, error) {
		return remoteCluster.DialAuthServer()
	}
}
868366
func (clstr *Cluster) clusterBoss(policy *ClientPolicy) {
	Logger.Info("Starting the cluster tend goroutine...")

	defer func() {
		if r := recover(); r != nil {
			Logger.Error("Cluster tend goroutine crashed: %s", debug.Stack())
			go clstr.clusterBoss(&clstr.clientPolicy)
		}
	}()

	defer clstr.wgTend.Done()

	tendInterval := policy.TendInterval
	if tendInterval <= 10*time.Millisecond {
		tendInterval = 10 * time.Millisecond
	}

Loop:
	for {
		select {
		case <-clstr.tendChannel:
			// tend channel closed
			Logger.Debug("Tend channel closed. Shutting down the cluster...")
			break Loop
		case <-time.After(tendInterval):
			tm := time.Now()
			if err := clstr.tend(); err != nil {
				Logger.Warn(err.Error())
			}

			// Tending took longer than requested tend interval.
			// Tending is too slow for the cluster, and may be falling behind scheule.
			if tendDuration := time.Since(tm); tendDuration > clstr.clientPolicy.TendInterval {
				Logger.Warn("Tending took %s, while your requested ClientPolicy.TendInterval is %s. Tends are slower than the interval, and may be falling behind the changes in the cluster.", tendDuration, clstr.clientPolicy.TendInterval)
			}
		}
	}

	// cleanup code goes here
	// close the nodes
	nodeArray := clstr.GetNodes()
	for _, node := range nodeArray {
		node.Close()
	}
}
654076
func (c *RESTClientset) ListClusters(options metav1.ListOptions) (*kops.ClusterList, error) {
	return c.KopsClient.Clusters(metav1.NamespaceAll).List(options)
}
405967
func (s *EnvoyApi) EnvoyClustersFromState() []*EnvoyCluster {
	clusters := make([]*EnvoyCluster, 0)

	s.state.RLock()
	defer s.state.RUnlock()

	svcs := s.state.ByService()
	for svcName, endpoints := range svcs {
		if len(endpoints) < 1 {
			continue
		}

		var svc *service.Service
		for _, endpoint := range endpoints {
			if endpoint.IsAlive() {
				svc = endpoint
				break
			}
		}

		if svc == nil {
			continue
		}

		for _, port := range svc.Ports {
			if port.ServicePort < 1 {
				continue
			}

			clusters = append(clusters, &EnvoyCluster{
				Name:             SvcName(svcName, port.ServicePort),
				Type:             "sds", // use Sidecar's SDS endpoint for the hosts
				ConnectTimeoutMs: 500,
				LBType:           "round_robin", // TODO figure this out!
				ServiceName:      SvcName(svcName, port.ServicePort),
			})
		}
	}

	return clusters
}
1343021
func (h *Handler) getClusters(w http.ResponseWriter, r *http.Request, p httprouter.Params, c *SessionContext) (interface{}, error) {
	resource, err := h.cfg.ProxyClient.GetClusterName()
	if err != nil {
		return nil, trace.Wrap(err)
	}

	response := ui.NewAvailableClusters(resource.GetClusterName(),
		h.cfg.Proxy.GetSites())

	return response, nil
}
336624
func NewInstanceIDHandler(svcs <-chan types.StorageService) types.Middleware {
	iidh := &instanceIDHandler{
		s2d: map[string]string{},
	}
	for s := range svcs {
		iidh.s2d[strings.ToLower(s.Name())] = strings.ToLower(s.Driver().Name())
	}
	return iidh
}
917477
func (cd CentroidDistance) ClusterDistance(c1, c2 *Cluster) float64 {
	return c1.Centroid.DistanceFrom(c2.Centroid)
}
67828
func (h *handler) handleSystemContainer(request *restful.Request, response *restful.Response) {
	query, err := parseStatsRequest(request)
	if err != nil {
		handleError(response, "/stats/container", err)
		return
	}

	// Non-Kubernetes container stats.
	containerName := path.Join("/", query.ContainerName)
	stats, err := h.provider.GetRawContainerInfo(
		containerName, query.cadvisorRequest(), query.Subcontainers)
	if err != nil {
		if _, ok := stats[containerName]; ok {
			// If the failure is partial, log it and return a best-effort response.
			klog.Errorf("Partial failure issuing GetRawContainerInfo(%v): %v", query, err)
		} else {
			handleError(response, fmt.Sprintf("/stats/container %v", query), err)
			return
		}
	}
	writeResponse(response, stats)
}
1096348
func (s *ClusterService) ReleaseDedicatedCluster(p *ReleaseDedicatedClusterParams) (*ReleaseDedicatedClusterResponse, error) {
	resp, err := s.cs.newRequest("releaseDedicatedCluster", p.toURLValues())
	if err != nil {
		return nil, err
	}

	var r ReleaseDedicatedClusterResponse
	if err := json.Unmarshal(resp, &r); err != nil {
		return nil, err
	}

	// If we have a async client, we need to wait for the async result
	if s.cs.async {
		b, err := s.cs.GetAsyncJobResult(r.JobID, s.cs.timeout)
		if err != nil {
			if err == AsyncTimeoutErr {
				return &r, err
			}
			return nil, err
		}

		if err := json.Unmarshal(b, &r); err != nil {
			return nil, err
		}
	}

	return &r, nil
}
327446
func (c *RESTClientset) UpdateCluster(cluster *kops.Cluster, status *kops.ClusterStatus) (*kops.Cluster, error) {
	glog.Warningf("validating cluster update client side; needs to move to server")
	old, err := c.GetCluster(cluster.Name)
	if err != nil {
		return nil, err
	}
	if err := validation.ValidateClusterUpdate(cluster, status, old).ToAggregate(); err != nil {
		return nil, err
	}

	namespace := restNamespaceForClusterName(cluster.Name)
	return c.KopsClient.Clusters(namespace).Update(cluster)
}
405449
func clusterSlotsParser(rd *proto.Reader, n int64) (interface{}, error) {
	slots := make([]ClusterSlot, n)
	for i := 0; i < len(slots); i++ {
		n, err := rd.ReadArrayLen()
		if err != nil {
			return nil, err
		}
		if n < 2 {
			err := fmt.Errorf("redis: got %d elements in cluster info, expected at least 2", n)
			return nil, err
		}

		start, err := rd.ReadIntReply()
		if err != nil {
			return nil, err
		}

		end, err := rd.ReadIntReply()
		if err != nil {
			return nil, err
		}

		nodes := make([]ClusterNode, n-2)
		for j := 0; j < len(nodes); j++ {
			n, err := rd.ReadArrayLen()
			if err != nil {
				return nil, err
			}
			if n != 2 && n != 3 {
				err := fmt.Errorf("got %d elements in cluster info address, expected 2 or 3", n)
				return nil, err
			}

			ip, err := rd.ReadString()
			if err != nil {
				return nil, err
			}

			port, err := rd.ReadString()
			if err != nil {
				return nil, err
			}

			nodes[j].Addr = net.JoinHostPort(ip, port)

			if n == 3 {
				id, err := rd.ReadString()
				if err != nil {
					return nil, err
				}
				nodes[j].Id = id
			}
		}

		slots[i] = ClusterSlot{
			Start: int(start),
			End:   int(end),
			Nodes: nodes,
		}
	}
	return slots, nil
}
1258105
func (c *Client) GetClusterStatus() (map[string]*clientv3.StatusResponse, error) {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   c.Endpoints,
		DialTimeout: 5 * time.Second,
		TLS:         c.TLS,
	})
	if err != nil {
		return nil, err
	}
	defer cli.Close()

	clusterStatus := make(map[string]*clientv3.StatusResponse)
	for _, ep := range c.Endpoints {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		resp, err := cli.Status(ctx, ep)
		cancel()
		if err != nil {
			return nil, err
		}
		clusterStatus[ep] = resp
	}
	return clusterStatus, nil
}
655265
func (db *db) GetCluster(ctx context.Context, server string) (*appv1.Cluster, error) {
	clusterSecret, err := db.getClusterSecret(server)
	if err != nil {
		if errorStatus, ok := status.FromError(err); ok && errorStatus.Code() == codes.NotFound && server == common.KubernetesInternalAPIServerAddr {
			return &localCluster, nil
		} else {
			return nil, err
		}
	}
	return secretToCluster(clusterSecret), nil
}
1074257
func (s *Server) HTTPHandler() http.Handler {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.httpHandler
}
1066942
func (c *Client) UpdateCluster(p *UpdateClusterParameter) (*Cluster, error) {
	obj, err := c.Request("updateCluster", convertParamToMap(p))
	if err != nil {
		return nil, err
	}
	return obj.(*Cluster), err
}