<commit_before>package meta

import (
	"bytes"
	crand "crypto/rand"
	"crypto/sha256"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"math/rand"
	"net"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"sync"
	"time"

	"github.com/gogo/protobuf/proto"
	"github.com/hashicorp/raft"
	"github.com/influxdb/influxdb"
	"github.com/influxdb/influxdb/influxql"
	"github.com/influxdb/influxdb/meta/internal"
	"golang.org/x/crypto/bcrypt"
)

// tcp.Mux header bytes.
const (
	MuxRaftHeader = 0
	MuxExecHeader = 1
	MuxRPCHeader  = 5

	// SaltBytes is the number of bytes used for salts
	SaltBytes = 32

	DefaultSyncNodeDelay = time.Second
)

// ExecMagic is the first 4 bytes sent to a remote exec connection to verify
// that it is coming from a remote exec client connection.
const ExecMagic = "EXEC"

// Retention policy settings.
const (
	AutoCreateRetentionPolicyName   = "default"
	AutoCreateRetentionPolicyPeriod = 0

	// MaxAutoCreatedRetentionPolicyReplicaN is the maximum replication factor that will
	// be set for auto-created retention policies.
	MaxAutoCreatedRetentionPolicyReplicaN = 3
)

// Raft configuration.
const (
	raftLogCacheSize      = 512
	raftSnapshotsRetained = 2
	raftTransportMaxPool  = 3
	raftTransportTimeout  = 10 * time.Second
	MaxRaftNodes          = 3
)

// Store represents a raft-backed metastore.
type Store struct {
	mu     sync.RWMutex
	path   string
	opened bool

	id uint64 // local node id

	// All peers in cluster. Used during bootstrapping.
	peers []string

	data *Data

	rpc *rpc

	// The address used by other nodes to reach this node.
	RemoteAddr net.Addr

	raftState raftState

	ready   chan struct{}
	err     chan error
	closing chan struct{}
	wg      sync.WaitGroup
	changed chan struct{}

	// clusterTracingEnabled controls whether low-level cluster communication is logged.
	// Useful for troubleshooting
	clusterTracingEnabled bool

	retentionAutoCreate bool

	// The listeners to accept raft and remote exec connections from.
	RaftListener net.Listener
	ExecListener net.Listener

	// The listener for higher-level, cluster operations
	RPCListener net.Listener

	// The advertised hostname of the store.
	Addr net.Addr

	// The amount of time before a follower starts a new election.
	HeartbeatTimeout time.Duration

	// The amount of time before a candidate starts a new election.
	ElectionTimeout time.Duration

	// The amount of time without communication to the cluster before a
	// leader steps down to a follower state.
	LeaderLeaseTimeout time.Duration

	// The amount of time without an apply before sending a heartbeat.
	CommitTimeout time.Duration

	// Authentication cache.
	authCache map[string]authUser

	// hashPassword generates a cryptographically secure hash for password.
	// Returns an error if the password is invalid or a hash cannot be generated.
	hashPassword HashPasswordFn

	// raftPromotionEnabled determines if non-raft nodes should be automatically
	// promoted to a raft node to self-heal a raft cluster
	raftPromotionEnabled bool

	Logger *log.Logger
}

type authUser struct {
	salt []byte
	hash []byte
}

// NewStore returns a new instance of Store.
func NewStore(c *Config) *Store {
	s := &Store{
		path:  c.Dir,
		peers: c.Peers,
		data:  &Data{},

		ready:   make(chan struct{}),
		err:     make(chan error),
		closing: make(chan struct{}),
		changed: make(chan struct{}),

		clusterTracingEnabled: c.ClusterTracing,
		retentionAutoCreate:   c.RetentionAutoCreate,
		raftPromotionEnabled:  c.RaftPromotionEnabled,

		HeartbeatTimeout:   time.Duration(c.HeartbeatTimeout),
		ElectionTimeout:    time.Duration(c.ElectionTimeout),
		LeaderLeaseTimeout: time.Duration(c.LeaderLeaseTimeout),
		CommitTimeout:      time.Duration(c.CommitTimeout),
		authCache:          make(map[string]authUser, 0),
		hashPassword: func(password string) ([]byte, error) {
			return bcrypt.GenerateFromPassword([]byte(password), BcryptCost)
		},
		Logger: log.New(os.Stderr, "[metastore] ", log.LstdFlags),
	}

	s.raftState = &localRaft{store: s}
	s.rpc = &rpc{
		store:          s,
		tracingEnabled: c.ClusterTracing,
		logger:         s.Logger,
	}
	return s
}
<commit_msg>silence meta/rpc logging for testing<commit_after>package meta

import (
	"bytes"
	crand "crypto/rand"
	"crypto/sha256"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"math/rand"
	"net"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"sync"
	"time"

	"github.com/gogo/protobuf/proto"
	"github.com/hashicorp/raft"
	"github.com/influxdb/influxdb"
	"github.com/influxdb/influxdb/influxql"
	"github.com/influxdb/influxdb/meta/internal"
	"golang.org/x/crypto/bcrypt"
)

// tcp.Mux header bytes.
const (
	MuxRaftHeader = 0
	MuxExecHeader = 1
	MuxRPCHeader  = 5

	// SaltBytes is the number of bytes used for salts
	SaltBytes = 32

	DefaultSyncNodeDelay = time.Second
)

// ExecMagic is the first 4 bytes sent to a remote exec connection to verify
// that it is coming from a remote exec client connection.
const ExecMagic = "EXEC"

// Retention policy settings.
const (
	AutoCreateRetentionPolicyName   = "default"
	AutoCreateRetentionPolicyPeriod = 0

	// MaxAutoCreatedRetentionPolicyReplicaN is the maximum replication factor that will
	// be set for auto-created retention policies.
	MaxAutoCreatedRetentionPolicyReplicaN = 3
)

// Raft configuration.
const (
	raftLogCacheSize      = 512
	raftSnapshotsRetained = 2
	raftTransportMaxPool  = 3
	raftTransportTimeout  = 10 * time.Second
	MaxRaftNodes          = 3
)

// Store represents a raft-backed metastore.
type Store struct {
	mu     sync.RWMutex
	path   string
	opened bool

	id uint64 // local node id

	// All peers in cluster. Used during bootstrapping.
	peers []string

	data *Data

	rpc *rpc

	// The address used by other nodes to reach this node.
	RemoteAddr net.Addr

	raftState raftState

	ready   chan struct{}
	err     chan error
	closing chan struct{}
	wg      sync.WaitGroup
	changed chan struct{}

	// clusterTracingEnabled controls whether low-level cluster communication is logged.
	// Useful for troubleshooting
	clusterTracingEnabled bool

	retentionAutoCreate bool

	// The listeners to accept raft and remote exec connections from.
	RaftListener net.Listener
	ExecListener net.Listener

	// The listener for higher-level, cluster operations
	RPCListener net.Listener

	// The advertised hostname of the store.
	Addr net.Addr

	// The amount of time before a follower starts a new election.
	HeartbeatTimeout time.Duration

	// The amount of time before a candidate starts a new election.
	ElectionTimeout time.Duration

	// The amount of time without communication to the cluster before a
	// leader steps down to a follower state.
	LeaderLeaseTimeout time.Duration

	// The amount of time without an apply before sending a heartbeat.
	CommitTimeout time.Duration

	// Authentication cache.
	authCache map[string]authUser

	// hashPassword generates a cryptographically secure hash for password.
	// Returns an error if the password is invalid or a hash cannot be generated.
	hashPassword HashPasswordFn

	// raftPromotionEnabled determines if non-raft nodes should be automatically
	// promoted to a raft node to self-heal a raft cluster
	raftPromotionEnabled bool

	Logger *log.Logger
}

type authUser struct {
	salt []byte
	hash []byte
}

// NewStore returns a new instance of Store.
func NewStore(c *Config) *Store {
	s := &Store{
		path:  c.Dir,
		peers: c.Peers,
		data:  &Data{},

		ready:   make(chan struct{}),
		err:     make(chan error),
		closing: make(chan struct{}),
		changed: make(chan struct{}),

		clusterTracingEnabled: c.ClusterTracing,
		retentionAutoCreate:   c.RetentionAutoCreate,
		raftPromotionEnabled:  c.RaftPromotionEnabled,

		HeartbeatTimeout:   time.Duration(c.HeartbeatTimeout),
		ElectionTimeout:    time.Duration(c.ElectionTimeout),
		LeaderLeaseTimeout: time.Duration(c.LeaderLeaseTimeout),
		CommitTimeout:      time.Duration(c.CommitTimeout),
		authCache:          make(map[string]authUser, 0),
		hashPassword: func(password string) ([]byte, error) {
			return bcrypt.GenerateFromPassword([]byte(password), BcryptCost)
		},
		Logger: log.New(os.Stderr, "[metastore] ", log.LstdFlags),
	}

	s.raftState = &localRaft{store: s}
	s.rpc = &rpc{
		store:          s,
		tracingEnabled: c.ClusterTracing,
		logger:         s.Logger,
	}
	return s
}

// SetLogger sets the internal logger to the logger passed in.
func (s *Store) SetLogger(l *log.Logger) {
	s.Logger = l
	if s.rpc != nil {
		s.rpc.logger = l
	}
}

{'fraction_non_alphanumeric': 0.08753691671588895, 'fraction_numerical': 0.005197873597164796, 'mean_word_length': 3.6929046563192904, 'pattern_counts': {'":': 0, '<': 3, '<?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': '1819475', 'n_tokens_mistral': 2867, 'n_tokens_neox': 2655, 'n_words': 846}