<commit_before>package main

import (
	"bytes"
	"compress/lzw"
	"crypto/md5"
	"encoding/gob"
	"fmt"
	"log"
	"path"
	"sync"
	"time"

	dep "github.com/hashicorp/consul-template/dependency"
	consulapi "github.com/hashicorp/consul/api"
)

const (
	// sessionCreateRetry is the amount of time we wait
	// to recreate a session when lost.
	sessionCreateRetry = 15 * time.Second

	// lockRetry is the interval on which we try to re-acquire locks
	lockRetry = 10 * time.Second

	// listRetry is the interval on which we retry listing a data path
	listRetry = 10 * time.Second

	// templateDataFlag is added as a flag to the shared data values
	// so that we can use it as a sanity check
	templateDataFlag = 0x22b9a127a2c03520
)

// templateData is GOB encoded share the depdency values
type templateData struct {
	Data map[string]interface{}
}

// DedupManager is used to de-duplicate which instance of Consul-Template
// is handling each template. For each template, a lock path is determined
// using the MD5 of the template. This path is used to elect a "leader"
// instance.
//
// The leader instance operations like usual, but any time a template is
// rendered, any of the data required for rendering is stored in the
// Consul KV store under the lock path.
//
// The follower instances depend on the leader to do the primary watching
// and rendering, and instead only watch the aggregated data in the KV.
// Followers wait for updates and re-render the template.
//
// If a template depends on 50 views, and is running on 50 machines, that
// would normally require 2500 blocking queries. Using deduplication, one
// instance has 50 view queries, plus 50 additional queries on the lock
// path for a total of 100.
//
type DedupManager struct {
	// config is the consul-template configuration
	config *Config

	// clients is used to access the underlying clinets
	clients *dep.ClientSet

	// Brain is where we inject udpates
	brain *Brain

	// templates is the set of templates we are trying to dedup
	templates []*Template

	// leader tracks if we are currently the leader
	leader     map[*Template]<-chan struct{}
	leaderLock sync.RWMutex

	// lastWrite tracks the hash of the data paths
	lastWrite     map[*Template][]byte
	lastWriteLock sync.RWMutex

	// updateCh is used to indicate an update watched data
	updateCh chan struct{}

	// wg is used to wait for a clean shutdown
	wg sync.WaitGroup

	stop     bool
	stopCh   chan struct{}
	stopLock sync.Mutex
}

// NewDedupManager creates a new Dedup manager
func NewDedupManager(config *Config, clients *dep.ClientSet, brain *Brain, templates []*Template) (*DedupManager, error) {
	d := &DedupManager{
		config:    config,
		clients:   clients,
		brain:     brain,
		templates: templates,
		leader:    make(map[*Template]<-chan struct{}),
		lastWrite: make(map[*Template][]byte),
		updateCh:  make(chan struct{}, 1),
		stopCh:    make(chan struct{}),
	}
	return d, nil
}

// Start is used to start the de-duplication manager
func (d *DedupManager) Start() error {
	log.Printf("[INFO] (dedup) starting de-duplication manager")

	client, err := d.clients.Consul()
	if err != nil {
		return err
	}
	go d.createSession(client)

	// Start to watch each template
	for _, t := range d.templates {
		go d.watchTemplate(client, t)
	}
	return nil
}

// Stop is used to stop the de-duplication manager
func (d *DedupManager) Stop() error {
	d.stopLock.Lock()
	defer d.stopLock.Unlock()
	if d.stop {
		return nil
	}

	log.Printf("[INFO] (dedup) stopping de-duplication manager")
	d.stop = true
	close(d.stopCh)
	d.wg.Wait()
	return nil
}

// createSession is used to create and maintain a session to Consul
func (d *DedupManager) createSession(client *consulapi.Client) {
START:
	log.Printf("[INFO] (dedup) attempting to create session")
	session := client.Session()
	sessionCh := make(chan struct{})
	ttl := fmt.Sprintf("%ds", d.config.Deduplicate.TTL/time.Second)
	se := &consulapi.SessionEntry{
		Name:     "Consul-Template de-duplication",
		Behavior: "delete",
		TTL:      ttl,
	}
	id, _, err := session.Create(se, nil)
	if err != nil {
		log.Printf("[ERR] (dedup) failed to create session: %v", err)
		goto WAIT
	}
	log.Printf("[INFO] (dedup) created session %s", id)

	// Attempt to lock each template
	for _, t := range d.templates {
		d.wg.Add(1)
		go d.attemptLock(client, id, sessionCh, t)
	}

	// Renew our session periodically
	if err := session.RenewPeriodic("15s", id, nil, d.stopCh); err != nil {
		log.Printf("[ERR] (dedup) failed to renew session: %v", err)
	}
	close(sessionCh)

WAIT:
	select {
	case <-time.After(sessionCreateRetry):
		goto START
	case <-d.stopCh:
		return
	}
}

// IsLeader checks if we are currently the leader instance
func (d *DedupManager) IsLeader(tmpl *Template) bool {
	d.leaderLock.RLock()
	defer d.leaderLock.RUnlock()

	lockCh, ok := d.leader[tmpl]
	if !ok {
		return false
	}
	select {
	case <-lockCh:<commit_msg>Wait for all existing dedupe acquire attempts to finish<commit_after>package main

import (
	"bytes"
	"compress/lzw"
	"crypto/md5"
	"encoding/gob"
	"fmt"
	"log"
	"path"
	"sync"
	"time"

	dep "github.com/hashicorp/consul-template/dependency"
	consulapi "github.com/hashicorp/consul/api"
)

const (
	// sessionCreateRetry is the amount of time we wait
	// to recreate a session when lost.
	sessionCreateRetry = 15 * time.Second

	// lockRetry is the interval on which we try to re-acquire locks
	lockRetry = 10 * time.Second

	// listRetry is the interval on which we retry listing a data path
	listRetry = 10 * time.Second

	// templateDataFlag is added as a flag to the shared data values
	// so that we can use it as a sanity check
	templateDataFlag = 0x22b9a127a2c03520
)

// templateData is GOB encoded share the depdency values
type templateData struct {
	Data map[string]interface{}
}

// DedupManager is used to de-duplicate which instance of Consul-Template
// is handling each template. For each template, a lock path is determined
// using the MD5 of the template. This path is used to elect a "leader"
// instance.
//
// The leader instance operations like usual, but any time a template is
// rendered, any of the data required for rendering is stored in the
// Consul KV store under the lock path.
//
// The follower instances depend on the leader to do the primary watching
// and rendering, and instead only watch the aggregated data in the KV.
// Followers wait for updates and re-render the template.
//
// If a template depends on 50 views, and is running on 50 machines, that
// would normally require 2500 blocking queries. Using deduplication, one
// instance has 50 view queries, plus 50 additional queries on the lock
// path for a total of 100.
//
type DedupManager struct {
	// config is the consul-template configuration
	config *Config

	// clients is used to access the underlying clinets
	clients *dep.ClientSet

	// Brain is where we inject udpates
	brain *Brain

	// templates is the set of templates we are trying to dedup
	templates []*Template

	// leader tracks if we are currently the leader
	leader     map[*Template]<-chan struct{}
	leaderLock sync.RWMutex

	// lastWrite tracks the hash of the data paths
	lastWrite     map[*Template][]byte
	lastWriteLock sync.RWMutex

	// updateCh is used to indicate an update watched data
	updateCh chan struct{}

	// wg is used to wait for a clean shutdown
	wg sync.WaitGroup

	stop     bool
	stopCh   chan struct{}
	stopLock sync.Mutex
}

// NewDedupManager creates a new Dedup manager
func NewDedupManager(config *Config, clients *dep.ClientSet, brain *Brain, templates []*Template) (*DedupManager, error) {
	d := &DedupManager{
		config:    config,
		clients:   clients,
		brain:     brain,
		templates: templates,
		leader:    make(map[*Template]<-chan struct{}),
		lastWrite: make(map[*Template][]byte),
		updateCh:  make(chan struct{}, 1),
		stopCh:    make(chan struct{}),
	}
	return d, nil
}

// Start is used to start the de-duplication manager
func (d *DedupManager) Start() error {
	log.Printf("[INFO] (dedup) starting de-duplication manager")

	client, err := d.clients.Consul()
	if err != nil {
		return err
	}
	go d.createSession(client)

	// Start to watch each template
	for _, t := range d.templates {
		go d.watchTemplate(client, t)
	}
	return nil
}

// Stop is used to stop the de-duplication manager
func (d *DedupManager) Stop() error {
	d.stopLock.Lock()
	defer d.stopLock.Unlock()
	if d.stop {
		return nil
	}

	log.Printf("[INFO] (dedup) stopping de-duplication manager")
	d.stop = true
	close(d.stopCh)
	d.wg.Wait()
	return nil
}

// createSession is used to create and maintain a session to Consul
func (d *DedupManager) createSession(client *consulapi.Client) {
START:
	log.Printf("[INFO] (dedup) attempting to create session")
	session := client.Session()
	sessionCh := make(chan struct{})
	ttl := fmt.Sprintf("%ds", d.config.Deduplicate.TTL/time.Second)
	se := &consulapi.SessionEntry{
		Name:     "Consul-Template de-duplication",
		Behavior: "delete",
		TTL:      ttl,
	}
	id, _, err := session.Create(se, nil)
	if err != nil {
		log.Printf("[ERR] (dedup) failed to create session: %v", err)
		goto WAIT
	}
	log.Printf("[INFO] (dedup) created session %s", id)

	// Attempt to lock each template
	for _, t := range d.templates {
		d.wg.Add(1)
		go d.attemptLock(client, id, sessionCh, t)
	}

	// Renew our session periodically
	if err := session.RenewPeriodic("15s", id, nil, d.stopCh); err != nil {
		log.Printf("[ERR] (dedup) failed to renew session: %v", err)
		d.wg.Wait()
	}
	close(sessionCh)

WAIT:
	select {
	case <-time.After(sessionCreateRetry):
		goto START
	case <-d.stopCh:
		return
	}
}

// IsLeader checks if we are currently the leader instance
func (d *DedupManager) IsLeader(tmpl *Template) bool {
	d.leaderLock.RLock()
	defer d.leaderLock.RUnlock()

	lockCh, ok := d.leader[tmpl]
	if !ok {
		return false
	}
	select {
	case <-lockCh:
{'fraction_non_alphanumeric': 0.1061067153580848, 'fraction_numerical': 0.008115236356258876, 'mean_word_length': 3.6156367041198503, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, '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': '2570406', 'n_tokens_mistral': 3359, 'n_tokens_neox': 3187, 'n_words': 1180}