<commit_before>// Copyright 2012, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package mysqlctl

import (
	"fmt"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"code.google.com/p/vitess/go/relog"
	"code.google.com/p/vitess/go/rpcwrap"
	"code.google.com/p/vitess/go/vt/dbconfigs"
)

/* API and config for UpdateStream Service */

const (
	DISABLED = iota
	ENABLED
)

type UpdateStream struct {
	mycnf          *Mycnf
	tabletType     string
	state          int32
	actionLock     sync.Mutex
	binlogPrefix   string
	logsDir        string
	usingRelayLogs bool
	logMetadata    *SlaveMetadata
	mysqld         *Mysqld
	clients        []*Blp
	stateWaitGroup sync.WaitGroup
}

type UpdateStreamRequest struct {
	StartPosition string
}

var UpdateStreamRpcService *UpdateStream

func RegisterUpdateStreamService(mycnf *Mycnf) {
	if UpdateStreamRpcService != nil {
		relog.Warning("Update Stream service already initialized")
		return
	}

	UpdateStreamRpcService = &UpdateStream{mycnf: mycnf}
	UpdateStreamRpcService.clients = make([]*Blp, 10)
	rpcwrap.RegisterAuthenticated(UpdateStreamRpcService)
}

func logError() {
	if x := recover(); x != nil {
		relog.Error("%s", x.(error).Error())
	}
}

func isReplicaType(tabletType string) bool {
	switch tabletType {
	case "replica", "rdonly", "batch":
		return true
	}
	return false
}

func dbcfgsCorrect(tabletType string, dbcfgs dbconfigs.DBConfigs) bool {
	switch tabletType {
	case "master":
		if dbcfgs.Dba.UnixSocket != "" {
			return true
		}
	case "replica", "rdonly", "batch":
		if dbcfgs.Dba.UnixSocket != "" && dbcfgs.Repl.UnixSocket != "" {
			return true
		}
	}
	return false
}

func EnableUpdateStreamService(tabletType string, dbcfgs dbconfigs.DBConfigs) {
	defer logError()
	updateStream := UpdateStreamRpcService
	updateStream.actionLock.Lock()
	defer updateStream.actionLock.Unlock()

	if !dbcfgsCorrect(tabletType, dbcfgs) {
		relog.Warning("missing/incomplete db configs file, cannot enable update stream service")
		return
	}

	if updateStream.isServiceEnabled() {
		disableUpdateStreamService()
	}

	atomic.StoreInt32(&updateStream.state, ENABLED)

	updateStream.mysqld = NewMysqld(updateStream.mycnf, dbcfgs.Dba, dbcfgs.Repl)
	updateStream.tabletType = tabletType
	updateStream.usingRelayLogs = false
	if tabletType == "master" {
		updateStream.binlogPrefix = updateStream.mycnf.BinLogPath
	} else if isReplicaType(tabletType) {
		updateStream.binlogPrefix = updateStream.mycnf.RelayLogPath
		updateStream.usingRelayLogs = true
	}
	updateStream.logsDir = path.Dir(updateStream.binlogPrefix)

	if updateStream.usingRelayLogs {
		updateStream.logMetadata = NewSlaveMetadata(updateStream.logsDir, updateStream.mycnf.RelayLogInfoPath)
		go updateStream.buildSlaveMetadata()
	}<commit_msg>squelch log spam while we wait for a proper fix<commit_after>// Copyright 2012, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package mysqlctl

import (
	"fmt"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"code.google.com/p/vitess/go/relog"
	"code.google.com/p/vitess/go/rpcwrap"
	"code.google.com/p/vitess/go/vt/dbconfigs"
)

/* API and config for UpdateStream Service */

const (
	DISABLED = iota
	ENABLED
)

type UpdateStream struct {
	mycnf          *Mycnf
	tabletType     string
	state          int32
	actionLock     sync.Mutex
	binlogPrefix   string
	logsDir        string
	usingRelayLogs bool
	logMetadata    *SlaveMetadata
	mysqld         *Mysqld
	clients        []*Blp
	stateWaitGroup sync.WaitGroup
}

type UpdateStreamRequest struct {
	StartPosition string
}

var UpdateStreamRpcService *UpdateStream

func RegisterUpdateStreamService(mycnf *Mycnf) {
	if UpdateStreamRpcService != nil {
		relog.Warning("Update Stream service already initialized")
		return
	}

	UpdateStreamRpcService = &UpdateStream{mycnf: mycnf}
	UpdateStreamRpcService.clients = make([]*Blp, 10)
	rpcwrap.RegisterAuthenticated(UpdateStreamRpcService)
}

func logError() {
	if x := recover(); x != nil {
		relog.Error("%s", x.(error).Error())
	}
}

func isReplicaType(tabletType string) bool {
	switch tabletType {
	case "replica", "rdonly", "batch":
		return true
	}
	return false
}

func dbcfgsCorrect(tabletType string, dbcfgs dbconfigs.DBConfigs) bool {
	switch tabletType {
	case "master":
		if dbcfgs.Dba.UnixSocket != "" {
			return true
		}
	case "replica", "rdonly", "batch":
		if dbcfgs.Dba.UnixSocket != "" && dbcfgs.Repl.UnixSocket != "" {
			return true
		}
	}
	return false
}

func EnableUpdateStreamService(tabletType string, dbcfgs dbconfigs.DBConfigs) {
	defer logError()
	updateStream := UpdateStreamRpcService
	updateStream.actionLock.Lock()
	defer updateStream.actionLock.Unlock()

	if !dbcfgsCorrect(tabletType, dbcfgs) {
		relog.Warning("missing/incomplete db configs file, cannot enable update stream service")
		return
	}

	if updateStream.isServiceEnabled() {
		disableUpdateStreamService()
	}

	atomic.StoreInt32(&updateStream.state, ENABLED)

	updateStream.mysqld = NewMysqld(updateStream.mycnf, dbcfgs.Dba, dbcfgs.Repl)
	updateStream.tabletType = tabletType
	updateStream.usingRelayLogs = false
	if tabletType == "master" {
		updateStream.binlogPrefix = updateStream.mycnf.BinLogPath
	} else if isReplicaType(tabletType) {
		updateStream.binlogPrefix = updateStream.mycnf.RelayLogPath
		updateStream.usingRelayLogs = true
	}
	updateStream.logsDir = path.Dir(updateStream.binlogPrefix)

	if updateStream.usingRelayLogs {
		updateStream.logMetadata = NewSlaveMetadata(updateStream.logsDir, updateStream.mycnf.RelayLogInfoPath)
		// FIXME(msolomon) commenting out for now per shrutip as we
		// have log spam from misconfiguration. To be removed properly
		// in a subsequent fix.
		// go updateStream.buildSlaveMetadata()
	}
{'fraction_non_alphanumeric': 0.09952445652173914, 'fraction_numerical': 0.0033967391304347825, 'mean_word_length': 4.392857142857143, 'pattern_counts': {'":': 6, '<': 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': '3202088', 'n_tokens_mistral': 2092, 'n_tokens_neox': 2023, 'n_words': 451}