<commit_before>package projector

import "fmt"
import "sync"
import "io"
import "os"
import "net/http"
import "strings"
import "encoding/json"
import "runtime/pprof"

import ap "github.com/couchbase/indexing/secondary/adminport"
import c "github.com/couchbase/indexing/secondary/common"
import projC "github.com/couchbase/indexing/secondary/projector/client"
import protobuf "github.com/couchbase/indexing/secondary/protobuf/projector"
import "code.google.com/p/goprotobuf/proto"
import "github.com/couchbase/indexing/secondary/logging"

// Projector data structure, a projector is connected to
// one or more upstream kv-nodes. Works in tandem with
// projector's adminport.
type Projector struct {
	admind ap.Server // admin-port server
	// lock protected fields.
	mu     sync.RWMutex
	topics map[string]*Feed // active topics
	config c.Config         // full configuration information.
	// immutable config params
	name        string // human readable name of the projector
	clusterAddr string // kv cluster's address to connect
	adminport   string // projector listens on this adminport
	maxvbs      int
	cpuProfFd   *os.File
	logPrefix   string
}

// NewProjector creates a news projector instance and
// starts a corresponding adminport.
func NewProjector(maxvbs int, config c.Config) *Projector {
	p := &Projector{
		topics: make(map[string]*Feed),
		maxvbs: maxvbs,
	}

	// Setup dynamic configuration propagation
	config, err := c.GetSettingsConfig(config)
	c.CrashOnError(err)

	pconfig := config.SectionConfig("projector.", true /*trim*/)
	p.name = pconfig["name"].String()
	p.clusterAddr = pconfig["clusterAddr"].String()
	p.adminport = pconfig["adminport.listenAddr"].String()
	ef := config["projector.routerEndpointFactory"]
	config["projector.routerEndpointFactory"] = ef

	p.config = config
	p.ResetConfig(config)

	p.logPrefix = fmt.Sprintf("PROJ[%s]", p.adminport)

	callb := func(cfg c.Config) {
		logging.Infof("%v settings notifier from metakv\n", p.logPrefix)
		cfg.LogConfig(p.logPrefix)
		p.ResetConfig(cfg)
	}
	c.SetupSettingsNotifier(callb, make(chan struct{}))

	cluster := p.clusterAddr
	if !strings.HasPrefix(p.clusterAddr, "http://") {
		cluster = "http://" + cluster
	}

	apConfig := config.SectionConfig("projector.adminport.", true)
	apConfig.SetValue("name", "PRAM")
	reqch := make(chan ap.Request)
	p.admind = ap.NewHTTPServer(apConfig, reqch)

	go p.mainAdminPort(reqch)
	go p.watcherDameon(config["projector.watchInterval"].Int())
	logging.Infof("%v started ...\n", p.logPrefix)
	return p
}

// GetConfig returns the config object from projector.
func (p *Projector) GetConfig() c.Config {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.config
}

// ResetConfig accepts a full-set or subset of global configuration
// and updates projector related fields.
func (p *Projector) ResetConfig(config c.Config) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// reset configuration.
	if cv, ok := config["projector.settings.log_level"]; ok {
		logging.SetLogLevel(logging.Level(cv.String()))
	}
	if cv, ok := config["projector.settings.log_override"]; ok {
		logging.AddOverride(cv.String())
	}
	if cv, ok := config["projector.maxCpuPercent"]; ok {
		c.SetNumCPUs(cv.Int())
	}
	p.config = p.config.Override(config)

	// CPU-profiling
	cpuProfFname := config["projector.cpuProfFname"].String()
	if config["projector.cpuProfile"].Bool() && p.cpuProfFd == nil {
		fmsg := "%v cpu profiling => %q\n"
		logging.Infof(fmsg, p.logPrefix, cpuProfFname)
		p.cpuProfFd = p.startCPUProfile(cpuProfFname)

	} else if config["projector.cpuProfile"].Bool() == false {
		if p.cpuProfFd != nil {
			pprof.StopCPUProfile()
			logging.Infof("%v cpu profiling stopped\n", p.logPrefix)
		}
		p.cpuProfFd = nil

	} else {
		logging.Warnf("%v cpu profiling already active !!\n", p.logPrefix)
	}

	// MEM-profiling
	memProfFname := config["projector.memProfFname"].String()
	if config["projector.memProfile"].Bool() {
		if p.takeMEMProfile(memProfFname) {
			fmsg := "%v mem profile => %q\n"
			logging.Infof(fmsg, p.logPrefix, memProfFname)
		}
	}
}

// GetFeedConfig from current configuration settings.
func (p *Projector) GetFeedConfig() c.Config {
	p.mu.Lock()
	defer p.mu.Unlock()

	config, _ := c.NewConfig(map[string]interface{}{})
	config["clusterAddr"] = p.config["clusterAddr"] // copy by value.
	pconfig := p.config.SectionConfig("projector.", true /*trim*/)
	for _, key := range FeedConfigParams() {
		config.Set(key, pconfig[key])
	}
	return config
}

// GetFeed object for `topic`.
// - return ErrorTopicMissing if topic is not started.
func (p *Projector) GetFeed(topic string) (*Feed, error) {
	p.mu.RLock()
	defer p.mu.RUnlock()
<commit_msg>projector: Do not assume dynamic setting has all fields<commit_after>package projector

import "fmt"
import "sync"
import "io"
import "os"
import "net/http"
import "strings"
import "encoding/json"
import "runtime/pprof"

import ap "github.com/couchbase/indexing/secondary/adminport"
import c "github.com/couchbase/indexing/secondary/common"
import projC "github.com/couchbase/indexing/secondary/projector/client"
import protobuf "github.com/couchbase/indexing/secondary/protobuf/projector"
import "code.google.com/p/goprotobuf/proto"
import "github.com/couchbase/indexing/secondary/logging"

// Projector data structure, a projector is connected to
// one or more upstream kv-nodes. Works in tandem with
// projector's adminport.
type Projector struct {
	admind ap.Server // admin-port server
	// lock protected fields.
	mu     sync.RWMutex
	topics map[string]*Feed // active topics
	config c.Config         // full configuration information.
	// immutable config params
	name        string // human readable name of the projector
	clusterAddr string // kv cluster's address to connect
	adminport   string // projector listens on this adminport
	maxvbs      int
	cpuProfFd   *os.File
	logPrefix   string
}

// NewProjector creates a news projector instance and
// starts a corresponding adminport.
func NewProjector(maxvbs int, config c.Config) *Projector {
	p := &Projector{
		topics: make(map[string]*Feed),
		maxvbs: maxvbs,
	}

	// Setup dynamic configuration propagation
	config, err := c.GetSettingsConfig(config)
	c.CrashOnError(err)

	pconfig := config.SectionConfig("projector.", true /*trim*/)
	p.name = pconfig["name"].String()
	p.clusterAddr = pconfig["clusterAddr"].String()
	p.adminport = pconfig["adminport.listenAddr"].String()
	ef := config["projector.routerEndpointFactory"]
	config["projector.routerEndpointFactory"] = ef

	p.config = config
	p.ResetConfig(config)

	p.logPrefix = fmt.Sprintf("PROJ[%s]", p.adminport)

	callb := func(cfg c.Config) {
		logging.Infof("%v settings notifier from metakv\n", p.logPrefix)
		cfg.LogConfig(p.logPrefix)
		p.ResetConfig(cfg)
	}
	c.SetupSettingsNotifier(callb, make(chan struct{}))

	cluster := p.clusterAddr
	if !strings.HasPrefix(p.clusterAddr, "http://") {
		cluster = "http://" + cluster
	}

	apConfig := config.SectionConfig("projector.adminport.", true)
	apConfig.SetValue("name", "PRAM")
	reqch := make(chan ap.Request)
	p.admind = ap.NewHTTPServer(apConfig, reqch)

	go p.mainAdminPort(reqch)
	go p.watcherDameon(config["projector.watchInterval"].Int())
	logging.Infof("%v started ...\n", p.logPrefix)
	return p
}

// GetConfig returns the config object from projector.
func (p *Projector) GetConfig() c.Config {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.config
}

// ResetConfig accepts a full-set or subset of global configuration
// and updates projector related fields.
func (p *Projector) ResetConfig(config c.Config) {
	p.mu.Lock()
	defer p.mu.Unlock()

	// reset configuration.
	if cv, ok := config["projector.settings.log_level"]; ok {
		logging.SetLogLevel(logging.Level(cv.String()))
	}
	if cv, ok := config["projector.settings.log_override"]; ok {
		logging.AddOverride(cv.String())
	}
	if cv, ok := config["projector.maxCpuPercent"]; ok {
		c.SetNumCPUs(cv.Int())
	}
	p.config = p.config.Override(config)

	// CPU-profiling
	cpuProfile, ok := config["projector.cpuProfile"]
	if ok && cpuProfile.Bool() && p.cpuProfFd == nil {
		cpuProfFname, ok := config["projector.cpuProfFname"]
		if ok {
			fname := cpuProfFname.String()
			logging.Infof("%v cpu profiling => %q\n", p.logPrefix, fname)
			p.cpuProfFd = p.startCPUProfile(fname)

		} else {
			logging.Errorf("Missing cpu-profile o/p filename\n")
		}

	} else if ok && !cpuProfile.Bool() {
		if p.cpuProfFd != nil {
			pprof.StopCPUProfile()
			logging.Infof("%v cpu profiling stopped\n", p.logPrefix)
		}
		p.cpuProfFd = nil

	} else {
		logging.Warnf("%v cpu profiling already active !!\n", p.logPrefix)
	}

	// MEM-profiling
	memProfile, ok := config["projector.memProfile"]
	if ok && memProfile.Bool() {
		memProfFname, ok := config["projector.memProfFname"]
		if ok {
			fname := memProfFname.String()
			if p.takeMEMProfile(fname) {
				logging.Infof("%v mem profile => %q\n", p.logPrefix, fname)
			}
		} else {
			logging.Errorf("Missing mem-profile o/p filename\n")
		}
	}
}

// GetFeedConfig from current configuration settings.
func (p *Projector) GetFeedConfig() c.Config {
	p.mu.Lock()
	defer p.mu.Unlock()

	config, _ := c.NewConfig(map[string]interface{}{})
	config["clusterAddr"] = p.config["clusterAddr"] // copy by value.
	pconfig := p.config.SectionConfig("projector.", true /*trim*/)
	for _, key := range FeedConfigParams() {
		config.Set(key, pconfig[key])
	}
	return config
}

// GetFeed object for `topic`.
// - return ErrorTopicMissing if topic is not started.
func (p *Projector) GetFeed(topic string) (*Feed, error) {
	p.mu.RLock()
	defer p.mu.RUnlock()

{'fraction_non_alphanumeric': 0.14223011808966454, 'fraction_numerical': 0.0, 'mean_word_length': 4.238095238095238, 'pattern_counts': {'":': 0, '<': 3, '<?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': '2341737', 'n_tokens_mistral': 3370, 'n_tokens_neox': 3206, 'n_words': 768}