// Returns a thread-compatible implementation of influxdb interactions.
436040
func newSink(c influxdb_common.InfluxdbConfig) core.DataSink {
	client, err := influxdb_common.NewClient(c)
	if err != nil {
		glog.Errorf("issues while creating an InfluxDB sink: %v, will retry on use", err)
	}
	return &influxdbSink{
		client:  client, // can be nil
		c:       c,
		conChan: make(chan struct{}, c.Concurrency),
	}
}
164668
func (s *Service) NextDueRun(ctx context.Context, taskID influxdb.ID) (int64, error) {
	var nextDue int64
	err := s.kv.View(ctx, func(tx Tx) error {
		due, err := s.nextDueRun(ctx, tx, taskID)
		if err != nil {
			return err
		}
		nextDue = due
		return nil
	})
	if err != nil {
		return 0, err
	}

	return nextDue, nil
}
958703
func (sink *influxdbSink) GetMetric(metricName string, metricKeys []core.HistoricalKey, start, end time.Time) (map[core.HistoricalKey][]core.TimestampedMetricValue, error) {
	for _, key := range metricKeys {
		if err := sink.checkSanitizedKey(&key); err != nil {
			return nil, err
		}
	}

	if err := sink.checkSanitizedMetricName(metricName); err != nil {
		return nil, err
	}

	query := sink.composeRawQuery(metricName, nil, metricKeys, start, end)

	sink.RLock()
	defer sink.RUnlock()

	resp, err := sink.runQuery(query)
	if err != nil {
		return nil, err
	}

	res := make(map[core.HistoricalKey][]core.TimestampedMetricValue, len(metricKeys))
	for i, key := range metricKeys {
		if len(resp[i].Series) < 1 {
			return nil, fmt.Errorf("No results for metric %q describing %q", metricName, key.String())
		}

		vals, err := sink.parseRawQueryRow(resp[i].Series[0])
		if err != nil {
			return nil, err
		}
		res[key] = vals
	}

	return res, nil
}
431285
func GetThreadInterpreter(th *starlark.Thread) *Interpreter {
	if intr := th.Local(threadIntrKey); intr != nil {
		return intr.(*Interpreter)
	}
	panic("not an Interpreter thread, no Interpreter in its locals")
}
449525
func (m *Mysql) gatherPerfEventWaits(db *sql.DB, serv string, acc telegraf.Accumulator) error {
	rows, err := db.Query(perfEventWaitsQuery)
	if err != nil {
		return err
	}
	defer rows.Close()

	var (
		event               string
		starCount, timeWait float64
	)

	servtag := getDSNTag(serv)
	tags := map[string]string{
		"server": servtag,
	}
	for rows.Next() {
		if err := rows.Scan(&event, &starCount, &timeWait); err != nil {
			return err
		}
		tags["event_name"] = event
		fields := map[string]interface{}{
			"events_waits_total":         starCount,
			"events_waits_seconds_total": timeWait / picoSeconds,
		}

		acc.AddFields("mysql_perf_schema", fields, tags)
	}
	return nil
}
95203
func (env *Env) SetHighPriorityBackgroundThreads(n int) {
	C.rocksdb_env_set_high_priority_background_threads(env.c, C.int(n))
}
134236
func NewMonitorArgs(database string, value interface{}, requests map[string]MonitorRequest) []interface{} {
	return []interface{}{database, value, requests}
}
537721
func InfluxDBWithTags(r metrics.Registry, d time.Duration, url, database, username, password, namespace string, tags map[string]string) {
	u, err := uurl.Parse(url)
	if err != nil {
		log.Warn("Unable to parse InfluxDB", "url", url, "err", err)
		return
	}

	rep := &reporter{
		reg:       r,
		interval:  d,
		url:       *u,
		database:  database,
		username:  username,
		password:  password,
		namespace: namespace,
		tags:      tags,
		cache:     make(map[string]int64),
	}
	if err := rep.makeClient(); err != nil {
		log.Warn("Unable to make InfluxDB client", "err", err)
		return
	}

	rep.run()
}
1191761
func (exp *explorerUI) SyncStatusPageIntercept(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if exp.ShowingSyncStatusPage() {
			exp.StatusPage(w, "Database Update Running. Please Wait.",
				"Blockchain sync is running. Please wait.", "", ExpStatusSyncing)
			return
		}
		// Otherwise, proceed to the next http handler.
		next.ServeHTTP(w, r)
	})
}
619051
func (l *HostStateListener) getExistingThreadStateIDs() []string {
	stateIds := make([]string, len(l.threads))
	i := 0
	for s := range l.threads {
		stateIds[i] = s
		i++
	}
	return stateIds
}
175059
func (c *Client) GetSecretKeys(ctx context.Context, orgID influxdb.ID) ([]string, error) {
	var vs []string
	err := c.db.View(func(tx *bolt.Tx) error {
		vals, err := c.getSecretKeys(ctx, tx, orgID)
		if err != nil {
			return err
		}

		vs = vals
		return nil
	})

	if err != nil {
		return nil, err
	}

	return vs, nil
}
723903
func pickCurrentThread(dbp Process, trapthread Thread, threads []Thread) error {
	for _, th := range threads {
		if bp := th.Breakpoint(); bp.Active && bp.Internal {
			return dbp.SwitchThread(th.ThreadID())
		}
	}
	if bp := trapthread.Breakpoint(); bp.Active {
		return dbp.SwitchThread(trapthread.ThreadID())
	}
	for _, th := range threads {
		if bp := th.Breakpoint(); bp.Active {
			return dbp.SwitchThread(th.ThreadID())
		}
	}
	return dbp.SwitchThread(trapthread.ThreadID())
}
827593
func (s *Semaphore) monitorLock(session string, stopCh chan struct{}) {
	defer close(stopCh)
	kv := s.c.KV()
	opts := &QueryOptions{RequireConsistent: true}
WAIT:
	retries := s.opts.MonitorRetries
RETRY:
	pairs, meta, err := kv.List(s.opts.Prefix, opts)
	if err != nil {
		// If configured we can try to ride out a brief Consul unavailability
		// by doing retries. Note that we have to attempt the retry in a non-
		// blocking fashion so that we have a clean place to reset the retry
		// counter if service is restored.
		if retries > 0 && IsRetryableError(err) {
			time.Sleep(s.opts.MonitorRetryTime)
			retries--
			opts.WaitIndex = 0
			goto RETRY
		}
		return
	}
	lockPair := s.findLock(pairs)
	lock, err := s.decodeLock(lockPair)
	if err != nil {
		return
	}
	s.pruneDeadHolders(lock, pairs)
	if _, ok := lock.Holders[session]; ok {
		opts.WaitIndex = meta.LastIndex
		goto WAIT
	}
}
775923
func (s *Service) FinishRun(ctx context.Context, taskID, runID influxdb.ID) (*influxdb.Run, error) {
	var run *influxdb.Run
	err := s.kv.Update(ctx, func(tx Tx) error {
		r, err := s.finishRun(ctx, tx, taskID, runID)
		if err != nil {
			return err
		}
		run = r
		return nil
	})
	return run, err
}
720551
func (c *EtcdLockClient) Get() (*Semaphore, error) {
	resp, err := c.keyapi.Get(context.Background(), c.keypath, nil)
	if err != nil {
		return nil, err
	}

	sem := &Semaphore{}
	err = json.Unmarshal([]byte(resp.Node.Value), sem)
	if err != nil {
		return nil, err
	}

	sem.Index = resp.Node.ModifiedIndex

	return sem, nil
}
932168
func (db *WiredDB) GetExplorerBlocks(start int, end int) []*exptypes.BlockBasic {
	if start < end {
		return nil
	}
	summaries := make([]*exptypes.BlockBasic, 0, start-end)
	for i := start; i > end; i-- {
		data := db.GetBlockVerbose(i, true)
		block := new(exptypes.BlockBasic)
		if data != nil {
			block = makeExplorerBlockBasic(data, db.params)
		}
		summaries = append(summaries, block)
	}
	return summaries
}
613338
func RegisterInfluxDB(ctx context.Context, config *types.InfluxDB) Registry {
	if influxDBClient == nil {
		influxDBClient = initInfluxDBClient(ctx, config)
	}
	if influxDBTicker == nil {
		influxDBTicker = initInfluxDBTicker(ctx, config)
	}

	return &standardRegistry{
		enabled:                        true,
		configReloadsCounter:           influxDBClient.NewCounter(influxDBConfigReloadsName),
		configReloadsFailureCounter:    influxDBClient.NewCounter(influxDBConfigReloadsFailureName),
		lastConfigReloadSuccessGauge:   influxDBClient.NewGauge(influxDBLastConfigReloadSuccessName),
		lastConfigReloadFailureGauge:   influxDBClient.NewGauge(influxDBLastConfigReloadFailureName),
		entrypointReqsCounter:          influxDBClient.NewCounter(influxDBEntrypointReqsName),
		entrypointReqDurationHistogram: influxDBClient.NewHistogram(influxDBEntrypointReqDurationName),
		entrypointOpenConnsGauge:       influxDBClient.NewGauge(influxDBEntrypointOpenConnsName),
		backendReqsCounter:             influxDBClient.NewCounter(influxDBMetricsBackendReqsName),
		backendReqDurationHistogram:    influxDBClient.NewHistogram(influxDBMetricsBackendLatencyName),
		backendRetriesCounter:          influxDBClient.NewCounter(influxDBRetriesTotalName),
		backendOpenConnsGauge:          influxDBClient.NewGauge(influxDBOpenConnsName),
		backendServerUpGauge:           influxDBClient.NewGauge(influxDBServerUpName),
	}
}