<commit_before>package server

import (
	"github.com/VividCortex/ewma"
	"github.com/klauspost/shutdown"
	"log"
	"net"
	"net/http"
	"sync"
	"time"
)

// A Backend is a single running backend instance.
// It will monitor itself and update health and stats every second.
type Backend interface {
	Transport() http.RoundTripper
	Host() string
	Healthy() bool
	Connections() int
	Close()
}

// backend is a common base used for sharing functionality
// between different backend types, so implementing different
// ones are easier.
type backend struct {
	rt           *statRT
	healthClient *http.Client
	closeMonitor chan chan struct{}
	Stats        Stats
	ServerHost   string
	HealthURL    string
}

// newBackend returns a new generic backend.
// It will start monitoring the backend at once
func newBackend(bec BackendConfig, serverHost, healthURL string) *backend {
	b := &backend{
		ServerHost: serverHost,
		HealthURL:  healthURL,
	}
	// Create a transport that is used for health checks.
	tr := &http.Transport{
		Dial: (&net.Dialer{
			Timeout:   time.Duration(bec.HealthTimeout),
			KeepAlive: 0,
		}).Dial,
		DisableKeepAlives:  true,
		DisableCompression: true,
	}
	b.healthClient = &http.Client{Transport: tr}

	// Reset running stats.
	b.Stats.Latency = ewma.NewMovingAverage(float64(bec.LatencyAvg))
	b.Stats.FailureRate = ewma.NewMovingAverage(10)

	// Set up the backend transport.
	tr = &http.Transport{
		Dial: func(network, addr string) (net.Conn, error) {
			return net.DialTimeout(network, addr, time.Duration(bec.DialTimeout))
		},
		Proxy: http.ProxyFromEnvironment,
	}
	b.rt = newStatTP(tr)

	// If we have no health url, assume healthy
	if healthURL == "" {
		b.Stats.Healthy = true
	}

	b.closeMonitor = make(chan chan struct{}, 0)

	go b.startMonitor()
	return b
}

// dropletBackend is a a backend instance with a DigitalOcean droplet
// behind it.
type dropletBackend struct {
	*backend
	Droplet Droplet
}

// NewDropletBackend returns a Backend configured with the
// Droplet information.
func NewDropletBackend(d Droplet, bec BackendConfig) Backend {
	b := &dropletBackend{
		backend: newBackend(bec, d.ServerHost, d.HealthURL),
		Droplet: d,
	}
	return b
}

// startMonitor will monitor stats of the backend
// Will at times require BOTH rt and Stats mutex.
// This means that no other goroutine should acquire
// both at the same time.
func (b *backend) startMonitor() {
	s := b.rt
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	exit := shutdown.First()
	end := b.closeMonitor
	previous := time.Now()

	for {
		select {
		case <-ticker.C:
			elapsed := time.Now().Sub(previous)
			previous = time.Now()
			s.mu.Lock()
			b.Stats.mu.Lock()
			if s.requests == 0 {
				b.Stats.Latency.Add(0)
				b.Stats.FailureRate.Add(0)
			} else {
				b.Stats.Latency.Add(float64(s.latencySum) / float64(elapsed) / float64(s.requests))
				b.Stats.FailureRate.Add(float64(s.errors) / float64(s.requests))
			}
			s.requests = 0
			s.errors = 0
			s.latencySum = 0
			s.mu.Unlock()

			// Perform health check
			b.healthCheck()

			if b.Stats.Healthy && b.Stats.healthFailures > 5 {
				log.Println("5 Consequtive health tests failed. Marking as unhealty.")
				b.Stats.Healthy = false
			}
			if !b.Stats.Healthy && b.Stats.healthFailures == 0 {
				log.Println("Health check succeeded. Marking as healty")
				b.Stats.Healthy = true
			}
			b.Stats.mu.Unlock()
		case n := <-end:
			exit.Cancel()
			close(n)
			return
		case n := <-exit:
			close(n)
			return
		}
	}
}

// healthCheck will check the health by connecting
// to the healthURL of the backend.
// This is called by healthCheck every second.
// It assumes b.Stats.mu is locked, but will unlock it while
// the request is running.
func (b *backend) healthCheck() {
	// If no checkurl har been set, assume we are healthy
	if b.HealthURL == "" {
		b.Stats.Healthy = true
		return
	}

	b.Stats.mu.Unlock()
	// Perform the check
	resp, err := b.healthClient.Get(b.HealthURL)

	b.Stats.mu.Lock()
	// Check response
	if err != nil {
		b.Stats.healthFailures++
		log.Println("Error checking health of", b.HealthURL, "Error:", err)
		return
	}
	if resp.StatusCode >= 500 {<commit_msg>Add custom User Agent to Health checks.<commit_after>package server

import (
	"github.com/VividCortex/ewma"
	"github.com/klauspost/shutdown"
	"log"
	"net"
	"net/http"
	"sync"
	"time"
)

// A Backend is a single running backend instance.
// It will monitor itself and update health and stats every second.
type Backend interface {
	Transport() http.RoundTripper
	Host() string
	Healthy() bool
	Connections() int
	Close()
}

// backend is a common base used for sharing functionality
// between different backend types, so implementing different
// ones are easier.
type backend struct {
	rt           *statRT
	healthClient *http.Client
	closeMonitor chan chan struct{}
	Stats        Stats
	ServerHost   string
	HealthURL    string
}

// newBackend returns a new generic backend.
// It will start monitoring the backend at once
func newBackend(bec BackendConfig, serverHost, healthURL string) *backend {
	b := &backend{
		ServerHost: serverHost,
		HealthURL:  healthURL,
	}
	// Create a transport that is used for health checks.
	tr := &http.Transport{
		Dial: (&net.Dialer{
			Timeout:   time.Duration(bec.HealthTimeout),
			KeepAlive: 0,
		}).Dial,
		DisableKeepAlives:  true,
		DisableCompression: true,
	}
	b.healthClient = &http.Client{Transport: tr}

	// Reset running stats.
	b.Stats.Latency = ewma.NewMovingAverage(float64(bec.LatencyAvg))
	b.Stats.FailureRate = ewma.NewMovingAverage(10)

	// Set up the backend transport.
	tr = &http.Transport{
		Dial: func(network, addr string) (net.Conn, error) {
			return net.DialTimeout(network, addr, time.Duration(bec.DialTimeout))
		},
		Proxy: http.ProxyFromEnvironment,
	}
	b.rt = newStatTP(tr)

	// If we have no health url, assume healthy
	if healthURL == "" {
		b.Stats.Healthy = true
	}

	b.closeMonitor = make(chan chan struct{}, 0)

	go b.startMonitor()
	return b
}

// dropletBackend is a a backend instance with a DigitalOcean droplet
// behind it.
type dropletBackend struct {
	*backend
	Droplet Droplet
}

// NewDropletBackend returns a Backend configured with the
// Droplet information.
func NewDropletBackend(d Droplet, bec BackendConfig) Backend {
	b := &dropletBackend{
		backend: newBackend(bec, d.ServerHost, d.HealthURL),
		Droplet: d,
	}
	return b
}

// startMonitor will monitor stats of the backend
// Will at times require BOTH rt and Stats mutex.
// This means that no other goroutine should acquire
// both at the same time.
func (b *backend) startMonitor() {
	s := b.rt
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	exit := shutdown.First()
	end := b.closeMonitor
	previous := time.Now()

	for {
		select {
		case <-ticker.C:
			elapsed := time.Now().Sub(previous)
			previous = time.Now()
			s.mu.Lock()
			b.Stats.mu.Lock()
			if s.requests == 0 {
				b.Stats.Latency.Add(0)
				b.Stats.FailureRate.Add(0)
			} else {
				b.Stats.Latency.Add(float64(s.latencySum) / float64(elapsed) / float64(s.requests))
				b.Stats.FailureRate.Add(float64(s.errors) / float64(s.requests))
			}
			s.requests = 0
			s.errors = 0
			s.latencySum = 0
			s.mu.Unlock()

			// Perform health check
			b.healthCheck()

			if b.Stats.Healthy && b.Stats.healthFailures > 5 {
				log.Println("5 Consequtive health tests failed. Marking as unhealty.")
				b.Stats.Healthy = false
			}
			if !b.Stats.Healthy && b.Stats.healthFailures == 0 {
				log.Println("Health check succeeded. Marking as healty")
				b.Stats.Healthy = true
			}
			b.Stats.mu.Unlock()
		case n := <-end:
			exit.Cancel()
			close(n)
			return
		case n := <-exit:
			close(n)
			return
		}
	}
}

// healthCheck will check the health by connecting
// to the healthURL of the backend.
// This is called by healthCheck every second.
// It assumes b.Stats.mu is locked, but will unlock it while
// the request is running.
func (b *backend) healthCheck() {
	// If no checkurl har been set, assume we are healthy
	if b.HealthURL == "" {
		b.Stats.Healthy = true
		return
	}

	req, err := http.NewRequest("GET", b.HealthURL, nil)
	if err != nil {
		log.Println("Error checking health of", b.HealthURL, "Error:", err)
	}

	req.Header.Set("User-Agent", "doproxy health checker")

	b.Stats.mu.Unlock()
	// Perform the check
	resp, err := b.healthClient.Do(req)

	b.Stats.mu.Lock()
	// Check response
	if err != nil {
		b.Stats.healthFailures++
		log.Println("Error checking health of", b.HealthURL, "Error:", err)
		return
	}
	if resp.StatusCode >= 500 {
{'fraction_non_alphanumeric': 0.12077633578861219, 'fraction_numerical': 0.0065474102654039515, 'mean_word_length': 3.3049823855057876, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4356303', 'n_tokens_mistral': 3180, 'n_tokens_neox': 2893, 'n_words': 853}