<commit_before>package consul

import (
	"bytes"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/hashicorp/consul/api"
	"github.com/portworx/kvdb"
)

const (
	// httpError is a substring returned by consul during such http errors.
	// Ideally such errors should be provided as consul constants
	httpError = "Unexpected response code: 500"
	// eofError is also a substring returned by consul during EOF errors.
	eofError = "EOF"
	// connRefused connection refused
	connRefused = "getsockopt: connection refused"
)

const (
	// keyIndexMismatch indicates consul error for key index mismatch
	keyIndexMismatch = "Key Index mismatch"
)

// clientConsul defines methods that a px based consul client should satisfy.
type consulClient interface {
	kvOperations
	// sessionOperations includes methods methods from that interface.
	sessionOperations
	// metaOperations includes methods from that interface.
	metaOperations
	// lockOptsOperations includes methods from that interface.
	lockOptsOperations
}

type kvOperations interface {
	// Get exposes underlying KV().Get but with reconnect on failover.
	Get(key string, q *api.QueryOptions) (*api.KVPair, *api.QueryMeta, error)
	// Put exposes underlying KV().Put but with reconnect on failover.
	Put(p *api.KVPair, q *api.WriteOptions) (*api.WriteMeta, error)
	// Acquire exposes underlying KV().Acquire but with reconnect on failover.
	Acquire(p *api.KVPair, q *api.WriteOptions) (*api.WriteMeta, error)
	// Delete exposes underlying KV().Delete but with reconnect on failover.
	Delete(key string, w *api.WriteOptions) (*api.WriteMeta, error)
	// DeleteTree exposes underlying KV().DeleteTree but with reconnect on failover.
	DeleteTree(prefix string, w *api.WriteOptions) (*api.WriteMeta, error)
	// Keys exposes underlying KV().Keys but with reconnect on failover.
	Keys(prefix, separator string, q *api.QueryOptions) ([]string, *api.QueryMeta, error)
	// List exposes underlying KV().List but with reconnect on failover.
	List(prefix string, q *api.QueryOptions) (api.KVPairs, *api.QueryMeta, error)
}

type sessionOperations interface {
	// Create exposes underlying Session().Create but with reconnect on failover.
	Create(se *api.SessionEntry, q *api.WriteOptions) (string, *api.WriteMeta, error)
	// Destroy exposes underlying Session().Destroy but with reconnect on failover.
	Destroy(id string, q *api.WriteOptions) (*api.WriteMeta, error)
	// Renew exposes underlying Session().Renew but with reconnect on failover.
	Renew(id string, q *api.WriteOptions) (*api.SessionEntry, *api.WriteMeta, error)
	// RenewPeriodic exposes underlying Session().RenewPeriodic but with reconnect on failover.
	RenewPeriodic(initialTTL string, id string, q *api.WriteOptions, doneCh chan struct{}) error
}

type metaOperations interface {
	// CreateMeta is a meta writer wrapping KV().Acquire and Session().Destroy but with reconnect on failover.
	CreateMeta(id string, p *api.KVPair, q *api.WriteOptions) (*api.WriteMeta, bool, error)
	// CompareAndSet is a meta func wrapping KV().CAS and KV().Get but with reconnect on failover.
	CompareAndSet(id string, value []byte, p *api.KVPair, q *api.WriteOptions) (bool, *api.WriteMeta, error)
	// CompareAndDelete is a meta func wrapping KV().DeleteCAS and KV().Get but with reconnect on failover.
	CompareAndDelete(id string, value []byte, p *api.KVPair, q *api.WriteOptions) (bool, *api.WriteMeta, error)
}

type lockOptsOperations interface {
	// LockOpts returns pointer to underlying Lock object and an error.
	LockOpts(opts *api.LockOptions) (*api.Lock, error)
}

// consulConnection stores current consul connection state
type consulConnection struct {
	// config is the configuration used to create consulClient
	config *api.Config
	// client provides access to consul api
	client *api.Client
	// once is used to reconnect consulClient only once among concurrently running threads
	once *sync.Once
}

// consulClient wraps config information and consul client along with sync functionality to reconnect it once.
// consulClient also satisfies interface defined above.
type consulClientImpl struct {
	// conn current consul connection state
	conn *consulConnection
	// connParams holds all params required to obtain new api client
	connParams connectionParams
	// reconnectDelay is the time duration to wait between machines
	reconnectDelay time.Duration
	// maxRetries is the number of times reconnect should be tried.
	maxRetries int
}

// newConsulClient provides an instance of clientConsul interface.
func newConsulClient(config *api.Config,
	client *api.Client,
	reconnectDelay time.Duration,
	p connectionParams,
) consulClient {
	c := &consulClientImpl{
		conn: &consulConnection{
			config: config,
			client: client,
			once:   new(sync.Once)},
		connParams:     p,
		reconnectDelay: reconnectDelay,
	}
	if len(c.connParams.machines) < 3 {
		c.maxRetries = 3
	} else {
		c.maxRetries = len(c.connParams.machines)
	}
	return c
}

// LockOpts returns pointer to underlying Lock object and an error.
func (c *consulClientImpl) LockOpts(opts *api.LockOptions) (*api.Lock, error) {
	return c.conn.client.LockOpts(opts)
}

// Refresh is PX specific op. It reconnectes consul client on failover.
func (c *consulClientImpl) reconnect(conn *consulConnection) error {
	var err error

	// once.Do executes func() only once across concurrently executing threads
	conn.once.Do(func() {
		var config *api.Config
		var client *api.Client

		for _, machine := range c.connParams.machines {
			if strings.HasPrefix(machine, "http://") {
				machine = strings.TrimPrefix(machine, "http://")
			} else if strings.HasPrefix(machine, "https://") {
				machine = strings.TrimPrefix(machine, "https://")
			}

			// sleep for requested delay before testing new connection
			time.Sleep(c.reconnectDelay)
			if config, client, err = newKvClient(machine, c.connParams); err == nil {
				c.conn = &consulConnection{
					client: client,
					config: config,
					once:   new(sync.Once),
				}
				logrus.Infof("%s: %s\n", "successfully connected to", machine)
				break
			} else {
				logrus.Errorf("failed to reconnect client on: %s", machine)
			}<commit_msg>More cleanup<commit_after>package consul

import (
	"bytes"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/hashicorp/consul/api"
	"github.com/portworx/kvdb"
)

const (
	// httpError is a substring returned by consul during such http errors.
	// Ideally such errors should be provided as consul constants
	httpError = "Unexpected response code: 500"
	// eofError is also a substring returned by consul during EOF errors.
	eofError = "EOF"
	// connRefused connection refused
	connRefused = "getsockopt: connection refused"
)

const (
	// keyIndexMismatch indicates consul error for key index mismatch
	keyIndexMismatch = "Key Index mismatch"
)

// clientConsul defines methods that a px based consul client should satisfy.
type consulClient interface {
	kvOperations
	// sessionOperations includes methods methods from that interface.
	sessionOperations
	// metaOperations includes methods from that interface.
	metaOperations
	// lockOptsOperations includes methods from that interface.
	lockOptsOperations
}

type kvOperations interface {
	// Get exposes underlying KV().Get but with reconnect on failover.
	Get(key string, q *api.QueryOptions) (*api.KVPair, *api.QueryMeta, error)
	// Put exposes underlying KV().Put but with reconnect on failover.
	Put(p *api.KVPair, q *api.WriteOptions) (*api.WriteMeta, error)
	// Acquire exposes underlying KV().Acquire but with reconnect on failover.
	Acquire(p *api.KVPair, q *api.WriteOptions) (*api.WriteMeta, error)
	// Delete exposes underlying KV().Delete but with reconnect on failover.
	Delete(key string, w *api.WriteOptions) (*api.WriteMeta, error)
	// DeleteTree exposes underlying KV().DeleteTree but with reconnect on failover.
	DeleteTree(prefix string, w *api.WriteOptions) (*api.WriteMeta, error)
	// Keys exposes underlying KV().Keys but with reconnect on failover.
	Keys(prefix, separator string, q *api.QueryOptions) ([]string, *api.QueryMeta, error)
	// List exposes underlying KV().List but with reconnect on failover.
	List(prefix string, q *api.QueryOptions) (api.KVPairs, *api.QueryMeta, error)
}

type sessionOperations interface {
	// Create exposes underlying Session().Create but with reconnect on failover.
	Create(se *api.SessionEntry, q *api.WriteOptions) (string, *api.WriteMeta, error)
	// Destroy exposes underlying Session().Destroy but with reconnect on failover.
	Destroy(id string, q *api.WriteOptions) (*api.WriteMeta, error)
	// Renew exposes underlying Session().Renew but with reconnect on failover.
	Renew(id string, q *api.WriteOptions) (*api.SessionEntry, *api.WriteMeta, error)
	// RenewPeriodic exposes underlying Session().RenewPeriodic but with reconnect on failover.
	RenewPeriodic(initialTTL string, id string, q *api.WriteOptions, doneCh chan struct{}) error
}

type metaOperations interface {
	// CreateMeta is a meta writer wrapping KV().Acquire and Session().Destroy but with reconnect on failover.
	CreateMeta(id string, p *api.KVPair, q *api.WriteOptions) (*api.WriteMeta, bool, error)
	// CompareAndSet is a meta func wrapping KV().CAS and KV().Get but with reconnect on failover.
	CompareAndSet(id string, value []byte, p *api.KVPair, q *api.WriteOptions) (bool, *api.WriteMeta, error)
	// CompareAndDelete is a meta func wrapping KV().DeleteCAS and KV().Get but with reconnect on failover.
	CompareAndDelete(id string, value []byte, p *api.KVPair, q *api.WriteOptions) (bool, *api.WriteMeta, error)
}

type lockOptsOperations interface {
	// LockOpts returns pointer to underlying Lock object and an error.
	LockOpts(opts *api.LockOptions) (*api.Lock, error)
}

// consulConnection stores current consul connection state
type consulConnection struct {
	// config is the configuration used to create consulClient
	config *api.Config
	// client provides access to consul api
	client *api.Client
	// once is used to reconnect consulClient only once among concurrently running threads
	once *sync.Once
}

// consulClient wraps config information and consul client along with sync functionality to reconnect it once.
// consulClient also satisfies interface defined above.
type consulClientImpl struct {
	// conn current consul connection state
	conn *consulConnection
	// connParams holds all params required to obtain new api client
	connParams connectionParams
	// reconnectDelay is the time duration to wait between machines
	reconnectDelay time.Duration
	// maxRetries is the number of times reconnect should be tried.
	maxRetries int
}

// newConsulClient provides an instance of clientConsul interface.
func newConsulClient(config *api.Config,
	client *api.Client,
	reconnectDelay time.Duration,
	p connectionParams,
) consulClient {
	c := &consulClientImpl{
		conn: &consulConnection{
			config: config,
			client: client,
			once:   new(sync.Once)},
		connParams:     p,
		reconnectDelay: reconnectDelay,
	}
	if len(c.connParams.machines) < 3 {
		c.maxRetries = 3
	} else {
		c.maxRetries = len(c.connParams.machines)
	}
	return c
}

// LockOpts returns pointer to underlying Lock object and an error.
func (c *consulClientImpl) LockOpts(opts *api.LockOptions) (*api.Lock, error) {
	return c.conn.client.LockOpts(opts)
}

// reconnect reconnectes to any online and healthy consul server..
func (c *consulClientImpl) reconnect(conn *consulConnection) error {
	var err error

	// once.Do executes func() only once across concurrently executing threads
	conn.once.Do(func() {
		var config *api.Config
		var client *api.Client

		for _, machine := range c.connParams.machines {
			if strings.HasPrefix(machine, "http://") {
				machine = strings.TrimPrefix(machine, "http://")
			} else if strings.HasPrefix(machine, "https://") {
				machine = strings.TrimPrefix(machine, "https://")
			}

			// sleep for requested delay before testing new connection
			time.Sleep(c.reconnectDelay)
			if config, client, err = newKvClient(machine, c.connParams); err == nil {
				c.conn = &consulConnection{
					client: client,
					config: config,
					once:   new(sync.Once),
				}
				logrus.Infof("%s: %s\n", "successfully connected to", machine)
				break
			} else {
				logrus.Errorf("failed to reconnect client on: %s", machine)
			}
{'fraction_non_alphanumeric': 0.09847748623258827, 'fraction_numerical': 0.0008098477486232588, 'mean_word_length': 4.35051993067591, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 4, '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': '2527197', 'n_tokens_mistral': 3877, 'n_tokens_neox': 3685, 'n_words': 1283}