Enhance agent startup logging, more generous startup restrictions (#196)

2.11.1 / CLDYCON-2913

Author: jdhudson3

charts/metrics-agent/Chart.yaml

@@ -14,8 +14,8 @@ type: application
 
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
-version: 2.11.0
+version: 2.11.1
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application.
-appVersion: 2.11.0
+appVersion: 2.11.1

charts/metrics-agent/values.yaml

@@ -19,7 +19,7 @@ pollInterval: 180
 
 image:
   name: cloudability/metrics-agent
-  tag: 2.11.0
+  tag: 2.11.1
   pullPolicy: Always
 
 imagePullSecrets: []

kubernetes/nodecollection.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"os"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/cloudability/metrics-agent/retrieval/raw"
@@ -29,7 +30,7 @@ func (err nodeError) Error() string {
 }
 
 const (
-	FatalNodeError = nodeError("unable to retrieve required set of node metrics via direct or proxy connection")
+	FatalNodeError = nodeError("unable to retrieve required metrics from any node via direct or proxy connection")
 )
 
 // NodeSource is an interface to get a list of Nodes
@@ -346,36 +347,87 @@ func ensureNodeSource(ctx context.Context, config KubeAgentConfig) (KubeAgentCon
 		return config, fmt.Errorf("error retrieving nodes: %s", err)
 	}
 
-	firstNode := &nodes[0]
+	directNodes := int32(0)
+	proxyNodes := int32(0)
+	failedDirect := int32(0)
+	failedProxy := int32(0)
+	directAllowed := allowDirectConnect(config, nodes)
 
-	ip, port, err := clientSetNodeSource.NodeAddress(firstNode)
-	if err != nil {
-		return config, fmt.Errorf("error retrieving node addresses: %s", err)
-	}
+	var wg sync.WaitGroup
 
-	if allowDirectConnect(config, nodes) {
-		// test node direct connectivity
-		d := directNodeEndpoints(ip, port)
-		success, err := checkEndpointConnections(config, &nodeHTTPClient, Direct, d.statsSummary())
-		if err != nil {
-			return config, err
-		}
-		if success {
-			return config, nil
-		}
+	limiter := make(chan struct{}, config.ConcurrentPollers)
+
+	for _, n := range nodes {
+		// block if channel is full (limiting number of goroutines)
+		limiter <- struct{}{}
+		wg.Add(1)
+		go func(currentNode v1.Node) {
+			defer func() {
+				<-limiter
+				wg.Done()
+			}()
+			directlyConnected := false
+			ip, port, err := clientSetNodeSource.NodeAddress(&currentNode)
+			if err != nil {
+				log.Warnf("error retrieving node addresses: %s", err)
+				return
+			}
+			if directAllowed {
+				// test node direct connectivity
+				d := directNodeEndpoints(ip, port)
+				success, err := checkEndpointConnections(config, &nodeHTTPClient, Direct, d.statsSummary())
+				if err != nil {
+					log.Warnf("Failed to connect to node [%s] directly with cause [%s]",
+						d.statsSummary(), err.Error())
+					atomic.AddInt32(&failedDirect, 1)
+				}
+				if success {
+					directlyConnected = true
+					atomic.AddInt32(&directNodes, 1)
+				}
+			}
+			if !directlyConnected {
+				p := setupProxyAPI(config.ClusterHostURL, currentNode.Name)
+				success, err := checkEndpointConnections(config, &config.HTTPClient, Proxy, p.statsSummary())
+				if err != nil {
+					log.Warnf("Failed to connect to node [%s] via proxy with cause [%s]",
+						p.statsSummary(), err.Error())
+					atomic.AddInt32(&failedProxy, 1)
+				}
+				if success {
+					atomic.AddInt32(&proxyNodes, 1)
+				}
+			}
+		}(n)
 	}
+	log.Debugln("Currently Waiting for all node data to be gathered")
+	wg.Wait()
+	log.Infof("Of %d nodes, %d connected directly, %d connected via proxy, and %d could not be reached",
+		len(nodes), directNodes, proxyNodes, failedProxy)
 
-	// test node connectivity via kube-proxy
-	p := setupProxyAPI(config.ClusterHostURL, firstNode.Name)
-	success, err := checkEndpointConnections(config, &config.HTTPClient, Proxy, p.statsSummary())
-	if err != nil {
-		return config, err
+	if len(nodes) != int(directNodes+proxyNodes) {
+		pct := int(directNodes+proxyNodes) * 100 / len(nodes)
+		log.Warnf("Only %d percent of ready nodes could could be connected to, "+
+			"agent will operate in a limited mode.", pct)
 	}
-	if success {
-		return config, nil
+
+	if (directNodes + proxyNodes) == 0 {
+		return config, FatalNodeError
 	}
 
-	return config, FatalNodeError
+	validateConfig(config, proxyNodes, directNodes)
+	return config, nil
+}
+
+func validateConfig(config KubeAgentConfig, proxyNodes, directNodes int32) {
+	if proxyNodes > 0 {
+		config.NodeMetrics.SetAvailability(NodeStatsSummaryEndpoint, Proxy, true)
+	} else if directNodes > 0 {
+		config.NodeMetrics.SetAvailability(NodeStatsSummaryEndpoint, Direct, true)
+	} else {
+		config.NodeMetrics.SetAvailability(NodeStatsSummaryEndpoint, Proxy, false)
+		config.NodeMetrics.SetAvailability(NodeStatsSummaryEndpoint, Direct, false)
+	}
 }
 
 func checkEndpointConnections(config KubeAgentConfig, client *http.Client, method Connection,
@@ -384,8 +436,7 @@ func checkEndpointConnections(config KubeAgentConfig, client *http.Client, metho
 	if err != nil {
 		return false, err
 	}
-	log.Infof("/stats/summary endpoint available via %s connection? %v", method, ns)
-	config.NodeMetrics.SetAvailability(NodeStatsSummaryEndpoint, method, ns)
+	log.Infof("Node [%s] available via %s connection? %v", nodeStatSum, method, ns)
 
 	return ns, nil
 }

kubernetes/nodecollection_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"k8s.io/apimachinery/pkg/runtime"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -121,13 +122,18 @@ var nodeSampleLabels = map[string]string{
 }
 
 func NewTestClient(ts *httptest.Server, labels map[string]string) *fake.Clientset {
+	return NewTestClientWithNodes(ts, labels, 1)
+}
+
+func NewTestClientWithNodes(ts *httptest.Server, labels map[string]string, numNodes int) *fake.Clientset {
 	s := strings.Split(ts.Listener.Addr().String(), ":")
 	ip := s[0]
 	port, _ := strconv.Atoi(s[1])
-	return fake.NewSimpleClientset(
-		&v1.Node{
+	nodes := make([]runtime.Object, numNodes)
+	for i := 0; i < numNodes; i++ {
+		nodes[i] = &v1.Node{
 			ObjectMeta: metav1.ObjectMeta{
-				Name:      "proxyNode",
+				Name:      fmt.Sprintf("proxyNode.%d", i),
 				Namespace: v1.NamespaceDefault,
 				Labels:    labels,
 			},
@@ -148,8 +154,9 @@ func NewTestClient(ts *httptest.Server, labels map[string]string) *fake.Clientse
 					},
 				},
 			},
-		},
-	)
+		}
+	}
+	return fake.NewSimpleClientset(nodes...)
 }
 
 func TestEnsureNodeSource(t *testing.T) {
@@ -185,6 +192,67 @@ func TestEnsureNodeSource(t *testing.T) {
 		}
 	})
 
+	t.Run("Ensure successful on mix of node failures and success", func(t *testing.T) {
+		returnCodes := []int{200, 400, 400}
+		ts := launchTLSTestServer(returnCodes)
+		cs := NewTestClientWithNodes(ts, nodeSampleLabels, 2)
+		defer ts.Close()
+		ka := KubeAgentConfig{
+			Clientset:            cs,
+			HTTPClient:           http.Client{},
+			CollectionRetryLimit: 0,
+			ConcurrentPollers:    10,
+			NodeMetrics:          EndpointMask{},
+		}
+		ka, err := ensureNodeSource(context.TODO(), ka)
+		if err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+		if !ka.NodeMetrics.DirectAllowed(NodeStatsSummaryEndpoint) {
+			t.Errorf("Expected direct node retrieval method but got %v: %v",
+				ka.NodeMetrics.Options(NodeStatsSummaryEndpoint),
+				err)
+			return
+		}
+	})
+
+	t.Run("Ensure proxy on mix of node direct and proxy", func(t *testing.T) {
+		returnCodes := []int{200, 400, 200}
+		ts := launchTLSTestServer(returnCodes)
+		cs := NewTestClientWithNodes(ts, nodeSampleLabels, 2)
+		defer ts.Close()
+		ka := KubeAgentConfig{
+			Clientset:            cs,
+			CollectionRetryLimit: 0,
+			ConcurrentPollers:    10,
+			NodeMetrics:          EndpointMask{},
+			ClusterHostURL:       "https://" + ts.Listener.Addr().String(),
+			// The proxy connection method uses the config http client
+			HTTPClient: http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{
+				// nolint gosec
+				InsecureSkipVerify: true,
+			},
+			}},
+		}
+		ka, err := ensureNodeSource(context.TODO(), ka)
+		if err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+		if ka.NodeMetrics.DirectAllowed(NodeStatsSummaryEndpoint) {
+			t.Errorf("Expected proxy node retrieval method but got %v: %v",
+				ka.NodeMetrics.Options(NodeStatsSummaryEndpoint),
+				err)
+			return
+		}
+
+		if !ka.NodeMetrics.ProxyAllowed(NodeStatsSummaryEndpoint) {
+			t.Errorf("Expected proxy node retrieval method but got %v: %v",
+				ka.NodeMetrics.Options(NodeStatsSummaryEndpoint),
+				err)
+			return
+		}
+	})
+
 	t.Run("Ensure all needed clients function when multiple methods are set", func(t *testing.T) {
 		// Two endpoints will succeed both times, but stats summary will fail on direct
 		directConnectionAttempts := []int{200}

version/version.go

@@ -1,4 +1,4 @@
 package version
 
 //VERSION is the current version of the agent
-var VERSION = "2.11.0"
+var VERSION = "2.11.1"