<commit_before>/*
Copyright 2016 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 agent

import (
	"fmt"
	"net"
	"runtime/debug"
	"sync"
	"time"

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

	log "github.com/Sirupsen/logrus"
	serf "github.com/hashicorp/serf/client"
	"github.com/jonboulle/clockwork"
	"golang.org/x/net/context"
)

// Agent is the interface to interact with the monitoring agent.
type Agent interface {
	// Start starts agent's background jobs.
	Start() error
	// Close stops background activity and releases resources.
	Close() error
	// Join makes an attempt to join a cluster specified by the list of peers.
	Join(peers []string) error
	// LocalStatus reports the health status of the local agent node.
	LocalStatus() *pb.NodeStatus

	health.CheckerRepository
}

type Config struct {
	// Name of the agent unique within the cluster.
	// Names are used as a unique id within a serf cluster, so
	// it is important to avoid clashes.
	//
	// Name must match the name of the local serf agent so that the agent
	// can match itself to a serf member.
	Name string

	// RPCAddrs is a list of addresses agent binds to for RPC traffic.
	//
	// Usually, at least two address are used for operation.
	// Localhost is a convenience for local communication.  Cluster-visible
	// IP is required for proper inter-communication between agents.
	RPCAddrs []string

	// RPC address of local serf node.
	SerfRPCAddr string

	// Peers lists the nodes that are part of the initial serf cluster configuration.
	// This is not a final cluster configuration and new nodes or node updates
	// are still possible.
	Peers []string

	// Set of tags for the agent.
	// Tags is a trivial means for adding extra semantic information to an agent.
	Tags map[string]string

	// Cache is a short-lived storage used by the agent to persist latest health stats.
	cache.Cache
}

// New creates an instance of an agent based on configuration options given in config.
func New(config *Config) (Agent, error) {
	clientConfig := &serf.Config{
		Addr: config.SerfRPCAddr,
	}
	client, err := serf.ClientFromConfig(clientConfig)
	if err != nil {
		return nil, trace.Wrap(err, "failed to connect to serf")
	}
	if config.Tags == nil {
		config.Tags = make(map[string]string)
	}
	err = client.UpdateTags(config.Tags, nil)
	if err != nil {
		return nil, trace.Wrap(err, "failed to update serf agent tags")
	}
	var listeners []net.Listener
	defer func() {
		if err != nil {
			for _, listener := range listeners {
				listener.Close()
			}
		}
	}()
	for _, addr := range config.RPCAddrs {
		listener, err := net.Listen("tcp", addr)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		listeners = append(listeners, listener)
	}
	clock := clockwork.NewRealClock()
	agent := &agent{
		serfClient:   client,
		name:         config.Name,
		cache:        config.Cache,
		dialRPC:      defaultDialRPC,
		statusClock:  clock,
		recycleClock: clock,
		localStatus:  emptyNodeStatus(config.Name),
	}
	agent.rpc = newRPCServer(agent, listeners)
	return agent, nil
}

type agent struct {
	health.Checkers

	// serfClient provides access to the serf agent.
	serfClient serfClient

	// Name of this agent.  Must be the same as the serf agent's name
	// running on the same node.
	name string

	// RPC server used by agent for client communication as well as
	// status sync with other agents.
	rpc RPCServer

	// cache persists node status history.
	cache cache.Cache

	// dialRPC is a factory function to create clients to other agents.
	// If future, agent address discovery will happen through serf.
	dialRPC dialRPC

	// done is a channel used for cleanup.
	done chan struct{}

	// These clocks abstract away access to the time package to allow
	// testing.
	statusClock  clockwork.Clock
	recycleClock clockwork.Clock

	mu sync.Mutex
	// localStatus is the last obtained local node status.
	localStatus *pb.NodeStatus
}

// Start starts the agent's background tasks.
func (r *agent) Start() error {
	r.done = make(chan struct{})

	go r.statusUpdateLoop()
	return nil
}

// Join attempts to join a serf cluster identified by peers.
func (r *agent) Join(peers []string) error {
	noReplay := false
	numJoined, err := r.serfClient.Join(peers, noReplay)
	if err != nil {
		return trace.Wrap(err)
	}
	log.Infof("joined %d nodes", numJoined)
	return nil
}

// Close stops all background activity and releases the agent's resources.
func (r *agent) Close() (err error) {
	r.rpc.Stop()
	close(r.done)
	err = r.serfClient.Close()
	if err != nil {
		return trace.Wrap(err)
	}
	return nil
}

// LocalStatus reports the status of the local agent node.
func (r *agent) LocalStatus() *pb.NodeStatus {
	return r.recentLocalStatus()
}

type dialRPC func(*serf.Member) (*client, error)

// runChecks executes the monitoring tests configured for this agent in parallel.
func (r *agent) runChecks(ctx context.Context) (*pb.NodeStatus, error) {
	// semaphoreCh limits the number of concurrent checkers
	semaphoreCh := make(chan struct{}, maxConcurrentCheckers)
	// channel for collecting resulting health probes
	probeCh := make(chan health.Probes, len(r.Checkers))

	for _, c := range r.Checkers {
		select {
		case semaphoreCh <- struct{}{}:
			go runChecker(ctx, c, probeCh, semaphoreCh)
		case <-ctx.Done():
			return nil, trace.ConnectionProblem(nil, "timed out running tests")
		}
	}

	var probes health.Probes
	for i := 0; i < len(r.Checkers); i++ {
		select {
		case probe := <-probeCh:
			probes = append(probes, probe...)
		case <-ctx.Done():
			return nil, trace.ConnectionProblem(nil, "timed out collecting test results")
		}
	}

	return &pb.NodeStatus{
		Name:   r.name,
		Status: probes.Status(),
		Probes: probes.GetProbes(),
	}, nil
}

// runChecker executes the specified checker and reports results on probeCh.
// If the checker panics, the resulting probe will describe the checker failure.
// Semaphore channel is guaranteed to receive a value upon completion.
func runChecker(ctx context.Context, checker health.Checker, probeCh chan<- health.Probes, semaphoreCh <-chan struct{}) {
	defer func() {
		if err := recover(); err != nil {
			var probes health.Probes
			probes.Add(&pb.Probe{
				Checker: checker.Name(),
				Status:  pb.Probe_Failed,
				Error:   trace.Errorf("checker panicked: %v\n%s", err, debug.Stack()).Error(),
			})
			probeCh <- probes
		}
		// release checker slot
		<-semaphoreCh
	}()

	log.Debugf("running checker %q", checker.Name())

	var probes health.Probes
	checker.Check(ctx, &probes)
	probeCh <- probes
}

// statusUpdateTimeout is the amount of time to wait between status update collections.
const statusUpdateTimeout = 30 * time.Second

// recycleTimeout is the amount of time to wait between recycle attempts.
// Recycle is a request to clean up / remove stale data that backends can choose to
// implement.
const recycleTimeout = 10 * time.Minute

// statusQueryReplyTimeout is the amount of time to wait for status query reply.
const statusQueryReplyTimeout = 30 * time.Second

// statusUpdateLoop is a long running background process that periodically
// updates the health status of the cluster by querying status of other active
// cluster members.
func (r *agent) statusUpdateLoop() {
	for {
		select {
		case <-r.statusClock.After(statusUpdateTimeout):
			ctx, cancel := context.WithTimeout(context.TODO(), statusQueryReplyTimeout)
			go func() {
				defer cancel() // close context if collection finishes before the deadline
				status, err := r.collectStatus(ctx)
				if err != nil {
					log.Warningf("error collecting system status: %v", err)
					return
				}
				if err = r.cache.UpdateStatus(status); err != nil {
					log.Warningf("error updating system status in cache: %v", err)
				}
			}()
			select {
			case <-ctx.Done():
				if ctx.Err() == context.DeadlineExceeded {
					log.Warningf("timed out collecting system status")
				}
			case <-r.done:
				cancel()
				return
			}
		case <-r.recycleClock.After(recycleTimeout):
			err := r.cache.Recycle()
			if err != nil {
				log.Warningf("error recycling stats: %v", err)
			}
		case <-r.done:
			return
		}
	}
}
<commit_msg>Fixed a bug with multiple time.After channels in a single select statement: reuse the previously created channels and reset only in case of handling.<commit_after>/*
Copyright 2016 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 agent

import (
	"fmt"
	"net"
	"runtime/debug"
	"sync"
	"time"

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

	log "github.com/Sirupsen/logrus"
	serf "github.com/hashicorp/serf/client"
	"github.com/jonboulle/clockwork"
	"golang.org/x/net/context"
)

// Agent is the interface to interact with the monitoring agent.
type Agent interface {
	// Start starts agent's background jobs.
	Start() error
	// Close stops background activity and releases resources.
	Close() error
	// Join makes an attempt to join a cluster specified by the list of peers.
	Join(peers []string) error
	// LocalStatus reports the health status of the local agent node.
	LocalStatus() *pb.NodeStatus

	health.CheckerRepository
}

type Config struct {
	// Name of the agent unique within the cluster.
	// Names are used as a unique id within a serf cluster, so
	// it is important to avoid clashes.
	//
	// Name must match the name of the local serf agent so that the agent
	// can match itself to a serf member.
	Name string

	// RPCAddrs is a list of addresses agent binds to for RPC traffic.
	//
	// Usually, at least two address are used for operation.
	// Localhost is a convenience for local communication.  Cluster-visible
	// IP is required for proper inter-communication between agents.
	RPCAddrs []string

	// RPC address of local serf node.
	SerfRPCAddr string

	// Peers lists the nodes that are part of the initial serf cluster configuration.
	// This is not a final cluster configuration and new nodes or node updates
	// are still possible.
	Peers []string

	// Set of tags for the agent.
	// Tags is a trivial means for adding extra semantic information to an agent.
	Tags map[string]string

	// Cache is a short-lived storage used by the agent to persist latest health stats.
	cache.Cache
}

// New creates an instance of an agent based on configuration options given in config.
func New(config *Config) (Agent, error) {
	clientConfig := &serf.Config{
		Addr: config.SerfRPCAddr,
	}
	client, err := serf.ClientFromConfig(clientConfig)
	if err != nil {
		return nil, trace.Wrap(err, "failed to connect to serf")
	}
	if config.Tags == nil {
		config.Tags = make(map[string]string)
	}
	err = client.UpdateTags(config.Tags, nil)
	if err != nil {
		return nil, trace.Wrap(err, "failed to update serf agent tags")
	}
	var listeners []net.Listener
	defer func() {
		if err != nil {
			for _, listener := range listeners {
				listener.Close()
			}
		}
	}()
	for _, addr := range config.RPCAddrs {
		listener, err := net.Listen("tcp", addr)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		listeners = append(listeners, listener)
	}
	clock := clockwork.NewRealClock()
	agent := &agent{
		serfClient:   client,
		name:         config.Name,
		cache:        config.Cache,
		dialRPC:      defaultDialRPC,
		statusClock:  clock,
		recycleClock: clock,
		localStatus:  emptyNodeStatus(config.Name),
	}
	agent.rpc = newRPCServer(agent, listeners)
	return agent, nil
}

type agent struct {
	health.Checkers

	// serfClient provides access to the serf agent.
	serfClient serfClient

	// Name of this agent.  Must be the same as the serf agent's name
	// running on the same node.
	name string

	// RPC server used by agent for client communication as well as
	// status sync with other agents.
	rpc RPCServer

	// cache persists node status history.
	cache cache.Cache

	// dialRPC is a factory function to create clients to other agents.
	// If future, agent address discovery will happen through serf.
	dialRPC dialRPC

	// done is a channel used for cleanup.
	done chan struct{}

	// These clocks abstract away access to the time package to allow
	// testing.
	statusClock  clockwork.Clock
	recycleClock clockwork.Clock

	mu sync.Mutex
	// localStatus is the last obtained local node status.
	localStatus *pb.NodeStatus
}

// Start starts the agent's background tasks.
func (r *agent) Start() error {
	r.done = make(chan struct{})

	go r.statusUpdateLoop()
	return nil
}

// Join attempts to join a serf cluster identified by peers.
func (r *agent) Join(peers []string) error {
	noReplay := false
	numJoined, err := r.serfClient.Join(peers, noReplay)
	if err != nil {
		return trace.Wrap(err)
	}
	log.Infof("joined %d nodes", numJoined)
	return nil
}

// Close stops all background activity and releases the agent's resources.
func (r *agent) Close() (err error) {
	r.rpc.Stop()
	close(r.done)
	err = r.serfClient.Close()
	if err != nil {
		return trace.Wrap(err)
	}
	return nil
}

// LocalStatus reports the status of the local agent node.
func (r *agent) LocalStatus() *pb.NodeStatus {
	return r.recentLocalStatus()
}

type dialRPC func(*serf.Member) (*client, error)

// runChecks executes the monitoring tests configured for this agent in parallel.
func (r *agent) runChecks(ctx context.Context) (*pb.NodeStatus, error) {
	// semaphoreCh limits the number of concurrent checkers
	semaphoreCh := make(chan struct{}, maxConcurrentCheckers)
	// channel for collecting resulting health probes
	probeCh := make(chan health.Probes, len(r.Checkers))

	for _, c := range r.Checkers {
		select {
		case semaphoreCh <- struct{}{}:
			go runChecker(ctx, c, probeCh, semaphoreCh)
		case <-ctx.Done():
			return nil, trace.ConnectionProblem(nil, "timed out running tests")
		}
	}

	var probes health.Probes
	for i := 0; i < len(r.Checkers); i++ {
		select {
		case probe := <-probeCh:
			probes = append(probes, probe...)
		case <-ctx.Done():
			return nil, trace.ConnectionProblem(nil, "timed out collecting test results")
		}
	}

	return &pb.NodeStatus{
		Name:   r.name,
		Status: probes.Status(),
		Probes: probes.GetProbes(),
	}, nil
}

// runChecker executes the specified checker and reports results on probeCh.
// If the checker panics, the resulting probe will describe the checker failure.
// Semaphore channel is guaranteed to receive a value upon completion.
func runChecker(ctx context.Context, checker health.Checker, probeCh chan<- health.Probes, semaphoreCh <-chan struct{}) {
	defer func() {
		if err := recover(); err != nil {
			var probes health.Probes
			probes.Add(&pb.Probe{
				Checker: checker.Name(),
				Status:  pb.Probe_Failed,
				Error:   trace.Errorf("checker panicked: %v\n%s", err, debug.Stack()).Error(),
			})
			probeCh <- probes
		}
		// release checker slot
		<-semaphoreCh
	}()

	log.Debugf("running checker %q", checker.Name())

	var probes health.Probes
	checker.Check(ctx, &probes)
	probeCh <- probes
}

// statusUpdateTimeout is the amount of time to wait between status update collections.
const statusUpdateTimeout = 30 * time.Second

// recycleTimeout is the amount of time to wait between recycle attempts.
// Recycle is a request to clean up / remove stale data that backends can choose to
// implement.
const recycleTimeout = 10 * time.Minute

// statusQueryReplyTimeout is the amount of time to wait for status query reply.
const statusQueryReplyTimeout = 30 * time.Second

// statusUpdateLoop is a long running background process that periodically
// updates the health status of the cluster by querying status of other active
// cluster members.
func (r *agent) statusUpdateLoop() {
	statusUpdateCh := r.statusClock.After(statusUpdateTimeout)
	recycleCh := r.recycleClock.After(recycleTimeout)
	for {
		select {
		case <-statusUpdateCh:
			ctx, cancel := context.WithTimeout(context.TODO(), statusQueryReplyTimeout)
			go func() {
				defer cancel() // close context if collection finishes before the deadline
				status, err := r.collectStatus(ctx)
				if err != nil {
					log.Warningf("error collecting system status: %v", err)
					return
				}
				if err = r.cache.UpdateStatus(status); err != nil {
					log.Warningf("error updating system status in cache: %v", err)
				}
			}()
			select {
			case <-ctx.Done():
				if ctx.Err() == context.DeadlineExceeded {
					log.Warningf("timed out collecting system status")
				}
			case <-r.done:
				cancel()
				return
			}
			statusUpdateCh = r.statusClock.After(statusUpdateTimeout)
		case <-recycleCh:
			err := r.cache.Recycle()
			if err != nil {
				log.Warningf("error recycling stats: %v", err)
			}
			recycleCh = r.recycleClock.After(recycleTimeout)
		case <-r.done:
			return
		}
	}
}

{'fraction_non_alphanumeric': 0.10009543591758828, 'fraction_numerical': 0.001684163251557851, 'mean_word_length': 3.553680981595092, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 3, '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': '4688344', 'n_tokens_mistral': 5819, 'n_tokens_neox': 5365, 'n_words': 2029}