<commit_before>/*
Copyright 2020 Gravitational, Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package monitoring

import (
	"bytes"
	"context"
	"fmt"
	"math"
	"net/url"
	"sync"
	"time"

	"github.com/gravitational/satellite/agent"
	"github.com/gravitational/satellite/agent/health"
	pb "github.com/gravitational/satellite/agent/proto/agentpb"
	"github.com/gravitational/satellite/utils"

	"github.com/gravitational/roundtrip"
	"github.com/gravitational/trace"
	"github.com/mailgun/holster"
	dto "github.com/prometheus/client_model/go"
	"github.com/prometheus/common/expfmt"
	log "github.com/sirupsen/logrus"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
)

// NethealthConfig specifies configuration for a nethealth checker.
type NethealthConfig struct {
	// NodeName specifies the kubernetes name of this node.
	NodeName string
	// NethealthPort specifies the port that nethealth is listening on.
	NethealthPort int
	// NetStatsInterval specifies the duration to store net stats.
	NetStatsInterval time.Duration
	// KubeConfig specifies kubernetes access information.
	*KubeConfig
}

// CheckAndSetDefaults validates that this configuration is correct and sets
// value defaults where necessary.
func (c *NethealthConfig) CheckAndSetDefaults() error {
	var errors []error
	if c.NodeName == "" {
		errors = append(errors, trace.BadParameter("node name must be provided"))
	}
	if c.KubeConfig == nil {
		errors = append(errors, trace.BadParameter("kubernetes access config must be provided"))
	}
	if c.NethealthPort == 0 {
		c.NethealthPort = defaultNethealthPort
	}
	if c.NetStatsInterval == time.Duration(0) {
		c.NetStatsInterval = defaultNetStatsInterval
	}
	return trace.NewAggregate(errors...)
}

// nethealthChecker checks network communication between peers.
type nethealthChecker struct {
	// NethealthConfig contains caller specified nethealth checker configuration
	// values.
	NethealthConfig
	// Mutex locks access to peerStats
	sync.Mutex
	// peerStats maps a peer's node name to its recorded nethealth stats.
	peerStats *netStats
}

// NewNethealthChecker returns a new nethealth checker.
func NewNethealthChecker(config NethealthConfig) (*nethealthChecker, error) {
	if err := config.CheckAndSetDefaults(); err != nil {
		return nil, trace.Wrap(err)
	}

	// seriesCapacity is determined by the NetStatsInterval / StatusUpdateTimeout
	// rounded up to the nearest integer
	seriesCapacity := math.Ceil(float64(config.NetStatsInterval) / float64(agent.StatusUpdateTimeout))

	return &nethealthChecker{
		peerStats:       newNetStats(netStatsCapacity, int(seriesCapacity)),
		NethealthConfig: config,
	}, nil
}

// Name returns this checker name
// Implements health.Checker
func (c *nethealthChecker) Name() string {
	return nethealthCheckerID
}

// Check verifies the network is healthy.
// Implements health.Checker
func (c *nethealthChecker) Check(ctx context.Context, reporter health.Reporter) {
	err := c.check(ctx, reporter)
	if err != nil {
		log.WithError(err).Warn("Failed to verify nethealth")
		reporter.Add(&pb.Probe{
			Checker:  c.Name(),
			Detail:   "failed to verify nethealth",
			Error:    trace.UserMessage(err),
			Status:   pb.Probe_Failed,
			Severity: pb.Probe_Warning,
		})
		return
	}
	if reporter.NumProbes() == 0 {
		reporter.Add(NewSuccessProbe(c.Name()))
	}
}

func (c *nethealthChecker) check(ctx context.Context, reporter health.Reporter) error {
	peers, err := c.getPeers()
	if err != nil {
		log.Debug("Failed to discover nethealth peers: %v.", err)
		return nil
	}
	if len(peers) == 0 {
		return nil
	}

	addr, err := c.getNethealthAddr()
	if trace.IsNotFound(err) {
		log.Debug("Nethealth pod was not found.")
		return nil // pod was not found, log and treat gracefully
	}
	if err != nil {
		return trace.Wrap(err) // received unexpected error, maybe network-related, will add error probe above
	}

	resp, err := fetchNethealthMetrics(ctx, addr)
	if err != nil {
		return trace.Wrap(err, "failed to fetch nethealth metrics")
	}<commit_msg>Debug -> Debugf (#212)<commit_after>/*
Copyright 2020 Gravitational, Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package monitoring

import (
	"bytes"
	"context"
	"fmt"
	"math"
	"net/url"
	"sync"
	"time"

	"github.com/gravitational/satellite/agent"
	"github.com/gravitational/satellite/agent/health"
	pb "github.com/gravitational/satellite/agent/proto/agentpb"
	"github.com/gravitational/satellite/utils"

	"github.com/gravitational/roundtrip"
	"github.com/gravitational/trace"
	"github.com/mailgun/holster"
	dto "github.com/prometheus/client_model/go"
	"github.com/prometheus/common/expfmt"
	log "github.com/sirupsen/logrus"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
)

// NethealthConfig specifies configuration for a nethealth checker.
type NethealthConfig struct {
	// NodeName specifies the kubernetes name of this node.
	NodeName string
	// NethealthPort specifies the port that nethealth is listening on.
	NethealthPort int
	// NetStatsInterval specifies the duration to store net stats.
	NetStatsInterval time.Duration
	// KubeConfig specifies kubernetes access information.
	*KubeConfig
}

// CheckAndSetDefaults validates that this configuration is correct and sets
// value defaults where necessary.
func (c *NethealthConfig) CheckAndSetDefaults() error {
	var errors []error
	if c.NodeName == "" {
		errors = append(errors, trace.BadParameter("node name must be provided"))
	}
	if c.KubeConfig == nil {
		errors = append(errors, trace.BadParameter("kubernetes access config must be provided"))
	}
	if c.NethealthPort == 0 {
		c.NethealthPort = defaultNethealthPort
	}
	if c.NetStatsInterval == time.Duration(0) {
		c.NetStatsInterval = defaultNetStatsInterval
	}
	return trace.NewAggregate(errors...)
}

// nethealthChecker checks network communication between peers.
type nethealthChecker struct {
	// NethealthConfig contains caller specified nethealth checker configuration
	// values.
	NethealthConfig
	// Mutex locks access to peerStats
	sync.Mutex
	// peerStats maps a peer's node name to its recorded nethealth stats.
	peerStats *netStats
}

// NewNethealthChecker returns a new nethealth checker.
func NewNethealthChecker(config NethealthConfig) (*nethealthChecker, error) {
	if err := config.CheckAndSetDefaults(); err != nil {
		return nil, trace.Wrap(err)
	}

	// seriesCapacity is determined by the NetStatsInterval / StatusUpdateTimeout
	// rounded up to the nearest integer
	seriesCapacity := math.Ceil(float64(config.NetStatsInterval) / float64(agent.StatusUpdateTimeout))

	return &nethealthChecker{
		peerStats:       newNetStats(netStatsCapacity, int(seriesCapacity)),
		NethealthConfig: config,
	}, nil
}

// Name returns this checker name
// Implements health.Checker
func (c *nethealthChecker) Name() string {
	return nethealthCheckerID
}

// Check verifies the network is healthy.
// Implements health.Checker
func (c *nethealthChecker) Check(ctx context.Context, reporter health.Reporter) {
	err := c.check(ctx, reporter)
	if err != nil {
		log.WithError(err).Warn("Failed to verify nethealth")
		reporter.Add(&pb.Probe{
			Checker:  c.Name(),
			Detail:   "failed to verify nethealth",
			Error:    trace.UserMessage(err),
			Status:   pb.Probe_Failed,
			Severity: pb.Probe_Warning,
		})
		return
	}
	if reporter.NumProbes() == 0 {
		reporter.Add(NewSuccessProbe(c.Name()))
	}
}

func (c *nethealthChecker) check(ctx context.Context, reporter health.Reporter) error {
	peers, err := c.getPeers()
	if err != nil {
		log.Debugf("Failed to discover nethealth peers: %v.", err)
		return nil
	}
	if len(peers) == 0 {
		return nil
	}

	addr, err := c.getNethealthAddr()
	if trace.IsNotFound(err) {
		log.Debug("Nethealth pod was not found.")
		return nil // pod was not found, log and treat gracefully
	}
	if err != nil {
		return trace.Wrap(err) // received unexpected error, maybe network-related, will add error probe above
	}

	resp, err := fetchNethealthMetrics(ctx, addr)
	if err != nil {
		return trace.Wrap(err, "failed to fetch nethealth metrics")
	}
{'fraction_non_alphanumeric': 0.10192265676207123, 'fraction_numerical': 0.005571334935547302, 'mean_word_length': 4.3695014662756595, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '886175', 'n_tokens_mistral': 2949, 'n_tokens_neox': 2845, 'n_words': 856}