func apiserverNode(ctx context.Context, nodes []Gravity) (api Gravity, other []Gravity, err error) {
	const apiserver = "leader.telekube.local"
	if len(nodes) < 1 {
		return nil, nil, trace.BadParameter("at least one node required")
	}
	apiserverAddr, err := ResolveInPlanet(ctx, nodes[0], apiserver)
	if err != nil {
		return nil, nil, trace.Wrap(err, "failed to resolve %v", apiserver)
	}
	other = make([]Gravity, 0, len(nodes)-1)
	for _, node := range nodes {
		if node.Node().PrivateAddr() == apiserverAddr {
			api = node
		} else {
			other = append(other, node)
		}
	}
	if api == nil {
		return nil, nil, trace.NotFound("no apiserver node found")
	}
	return api, other, nil
}
apiserverNode returns the node that runs the API server
package gravity
import (
	"bufio"
	"context"
	"encoding/json"
	"github.com/cenkalti/backoff"
	"golang.org/x/sync/errgroup"
	sshutils "github.com/gravitational/robotest/lib/ssh"
	"github.com/gravitational/robotest/lib/utils"
	"github.com/gravitational/robotest/lib/wait"
	"github.com/gravitational/trace"
)
const (
	ClusterStateActive = "active"
	ClusterStateDegraded = "degraded"
)
type SystemStatus int
const (
	SystemStatus_Unknown SystemStatus = 0
	SystemStatus_Running SystemStatus = 1
	SystemStatus_Degraded SystemStatus = 2
)
var systemStatusNames = [...]string{"undefined", "running", "degraded"}
func (ss SystemStatus) String() string {
	return systemStatusNames[ss]
}
func (s *SystemStatus) UnmarshalJSON(data []byte) error {
	var intStatus int
	err := json.Unmarshal(data, &intStatus)
	if err == nil {
		*s = SystemStatus(intStatus)
		return nil
	}
	var strStatus string
	if err2 := json.Unmarshal(data, &strStatus); err2 != nil {
		return trace.BadParameter("unable to parse system_status %q as int or string: %v, %v", data, err, err2)
	}
	switch strStatus {
	case SystemStatus_Running.String():
		*s = SystemStatus_Running
	case SystemStatus_Degraded.String():
		*s = SystemStatus_Degraded
	case SystemStatus_Unknown.String():
		*s = SystemStatus_Unknown
	default:
		return trace.BadParameter("unknown system_status: %q", strStatus)
	}
	return nil
}
func parseStatus(status *GravityStatus) sshutils.OutputParseFn {
	return func(r *bufio.Reader) error {
		decoder := json.NewDecoder(r)
		return trace.Wrap(decoder.Decode(status))
	}
}
type GravityStatus struct {
	Cluster ClusterStatus `json:"cluster"`
}
type ClusterStatus struct {
	Application Application `json:"application"`
	Cluster string `json:"domain"`
	State string `json:"state"`
	SystemStatus SystemStatus `json:"system_status"`
	Token Token `json:"token"`
	Nodes []NodeStatus `json:"nodes"`
}
type Application struct {
	Name string `json:"name"`
}
type NodeStatus struct {
	Addr string `json:"advertise_ip"`
}
type Token struct {
	Token string `json:"token"`
}
type statusValidator func(s GravityStatus) error
func checkNotDegraded(s GravityStatus) error {
	if s.Cluster.State == ClusterStateDegraded {
		return trace.CompareFailed("cluster state %q", s.Cluster.State)
	}
	if s.Cluster.SystemStatus != SystemStatus_Running {
		return trace.CompareFailed("expected system_status %v, found %v", SystemStatus_Running, s.Cluster.SystemStatus)
	}
	return nil
}
func checkActive(s GravityStatus) error {
	if err := checkNotDegraded(s); err != nil {
		return trace.Wrap(err)
	}
	if s.Cluster.State != ClusterStateActive {
		return trace.CompareFailed("expected state %q, found %q", ClusterStateActive, s.Cluster.State)
	}
	return nil
}
func (c *TestContext) WaitForActiveStatus(nodes []Gravity) error {
	c.Logger().WithField("nodes", Nodes(nodes)).Info("Waiting for active status.")
	return c.WaitForStatus(nodes, checkActive)
}
func (c *TestContext) WaitForStatus(nodes []Gravity, expected statusValidator) error {
	b := backoff.NewExponentialBackOff()
	b.MaxElapsedTime = c.timeouts.ClusterStatus
	expectStatus := func() (err error) {
		statuses, err := c.Status(nodes)
		if err != nil {
			return trace.Wrap(err)
		}
		for _, status := range statuses {
			err = expected(status)
			if err != nil {
				c.Logger().WithError(err).WithField("status", status).Warn("Unexpected Status.")
				return trace.Wrap(err)
			}
		}
		return nil
	}
	err := wait.RetryWithInterval(c.ctx, b, expectStatus, c.Logger())
	return trace.Wrap(err)
}
func (c *TestContext) Status(nodes []Gravity) (statuses []GravityStatus, err error) {
	ctx, cancel := context.WithTimeout(c.ctx, c.timeouts.NodeStatus)
	defer cancel()
	valueC := make(chan GravityStatus, len(nodes))
	g, ctx := errgroup.WithContext(ctx)
	for _, node := range nodes {
		node := node
		g.Go(func() error {
			status, err := node.Status(ctx)
			if err != nil {
				return trace.Wrap(err)
			}
			if status != nil {
				valueC <- *status
			}
			return nil
		})
	}
	err = g.Wait()
	if err != nil {
		return nil, trace.Wrap(err)
	}
	close(valueC)
	for status := range valueC {
		statuses = append(statuses, status)
	}
	return statuses, nil
}
func (c *TestContext) CheckTimeSync(nodes []Gravity) error {
	timeNodes := []sshutils.SshNode{}
	for _, n := range nodes {
		timeNodes = append(timeNodes, sshutils.SshNode{
			Client: n.Client(),
			Log:    c.Logger(),
		})
	}
	ctx, cancel := context.WithTimeout(c.ctx, c.timeouts.TimeSync)
	defer cancel()
	err := sshutils.CheckTimeSync(ctx, timeNodes)
	return trace.Wrap(err)
}
func (c *TestContext) CollectLogs(prefix string, nodes []Gravity) error {
	if len(nodes) < 1 {
		return nil
	}
	ctx, cancel := context.WithTimeout(c.ctx, c.timeouts.CollectLogs)
	defer cancel()
	c.Logger().WithField("nodes", nodes).Debug("Collecting logs from nodes.")
	api, other, err := apiserverNode(ctx, nodes)
	if err != nil || api == nil {
		c.Logger().WithError(err).Warn("Unable to determine api-server.")
	}
	errors := make(chan error, len(nodes))
	var args []string
	if api != nil {
		nodes = other  
		args = []string{"--filter=system", "--filter=kubernetes"}
		go collectLogsFromNode(ctx, api, prefix, args, errors)
	}
	args = []string{"--filter=system"}
	for _, node := range nodes {
		go collectLogsFromNode(ctx, node, prefix, args, errors)
	}
	err = utils.CollectErrors(ctx, errors)
	return trace.Wrap(err)
}
func collectLogsFromNode(ctx context.Context, node Gravity, prefix string, args []string, errors chan<- error) {
	node.Logger().Debug("Fetching node logs.")
	localPath, err := node.CollectLogs(ctx, prefix, args...)
	if err != nil {
		node.Logger().WithError(err).Error("Log fetch failed.")
	} else {
		node.Logger().WithField("path", localPath).Info("Logs saved.")
	}
	errors <- err
}
type ClusterNodesByRole struct {
	ApiMaster Gravity
	ClusterMaster Gravity
	ClusterBackup []Gravity
	Regular []Gravity
	Other []Gravity
}
func (c *TestContext) NodesByRole(nodes []Gravity) (roles *ClusterNodesByRole, err error) {
	ctx, cancel := context.WithTimeout(c.ctx, c.timeouts.ResolveInPlanet)
	defer cancel()
	roles = &ClusterNodesByRole{}
	roles.ApiMaster, roles.Other, err = apiserverNode(ctx, nodes)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	ctx, cancel = context.WithTimeout(c.ctx, c.timeouts.GetPods)
	defer cancel()
	pods, err := KubectlGetPods(ctx, roles.ApiMaster, kubeSystemNS, appGravityLabel)
	if err != nil {
		return nil, trace.Wrap(err)
	}
L:
	for _, node := range nodes {
		ip := node.Node().PrivateAddr()
		for _, pod := range pods {
			if ip == pod.NodeIP {
				if pod.Ready {
					roles.ClusterMaster = node
				} else {
					roles.ClusterBackup = append(roles.ClusterBackup, node)
				}
				continue L
			}
		}
		roles.Regular = append(roles.Regular, node)
	}
	return roles, nil
}