// NewKafkaPartitionConsumerMetrics creates new KafkaPartitionConsumerMetrics for a given topic and partition.
517875
func NewKafkaPartitionConsumerMetrics(topic string, partition int32) *KafkaPartitionConsumerMetrics {
	registry := metrics.NewPrefixedRegistry(fmt.Sprintf("%s.%d.", topic, partition))

	return &KafkaPartitionConsumerMetrics{
		registry:               registry,
		batchDuration:          metrics.NewRegisteredTimer("batchDuration", registry),
		fetchDuration:          metrics.NewRegisteredTimer("fetchDuration", registry),
		numFetches:             metrics.NewRegisteredCounter("numFetches", registry),
		numFailedFetches:       metrics.NewRegisteredCounter("numFailedFetches", registry),
		numEmptyFetches:        metrics.NewRegisteredCounter("numEmptyFetches", registry),
		numFetchedMessages:     metrics.NewRegisteredCounter("numFetchedMessages", registry),
		numOffsetCommits:       metrics.NewRegisteredCounter("numOffsetCommits", registry),
		numFailedOffsetCommits: metrics.NewRegisteredCounter("numFailedOffsetCommits", registry),
		lag: metrics.NewRegisteredGauge("lag", registry),
	}
}
571357
func NewProtoConsumerMessage(msg *ConsumerMessage, serializer keyval.Serializer) *ProtoConsumerMessage {
	return &ProtoConsumerMessage{msg, serializer}
}
461626
func newPodMetricses(c *MetricsV1beta1Client, namespace string) *podMetricses {
	return &podMetricses{
		client: c.RESTClient(),
		ns:     namespace,
	}
}
730695
func (tc *WildcardTopicsToNumStreams) GetConsumerThreadIdsPerTopic() map[string][]ConsumerThreadId {
	topicsToNumStreams := make(map[string]int)
	topics, err := tc.Coordinator.GetAllTopics()
	if err != nil {
		panic(err)
	}
	for _, topic := range topics {
		if tc.TopicFilter.TopicAllowed(topic, tc.ExcludeInternalTopics) {
			topicsToNumStreams[topic] = tc.NumStreams
		}
	}
	return makeConsumerThreadIdsPerTopic(tc.ConsumerId, topicsToNumStreams)
}
334047
func (cg *ConsumerGroup) partitionConsumer(topic string, partition int32, messages chan<- *sarama.ConsumerMessage, errors chan<- error, wg *sync.WaitGroup, stopper <-chan struct{}) {
	defer wg.Done()

	// Since ProcessingTimeout is the amount of time we'll wait for the final batch
	// of messages to be processed before releasing a partition, we need to wait slightly
	// longer than that before timing out here to ensure that another consumer has had
	// enough time to release the partition. Hence, +2 seconds.
	maxRetries := int(cg.config.Offsets.ProcessingTimeout/time.Second) + 2
partitionClaimLoop:
	for tries := 0; tries < maxRetries; tries++ {
		select {
		case <-stopper:
			return
		case <-time.After(1 * time.Second):
			if err := cg.instance.ClaimPartition(topic, partition); err == nil {
				break partitionClaimLoop
			} else if tries+1 < maxRetries {
				if err == kazoo.ErrPartitionClaimedByOther {
					// Another consumer still owns this partition. We should wait longer for it to release it.
				} else {
					// An unexpected error occurred. Log it and continue trying until we hit the timeout.
					cg.Logf("%s/%d :: FAILED to claim partition on attempt %v of %v; retrying in 1 second. Error: %v", topic, partition, tries+1, maxRetries, err)
				}
			} else {
				cg.Logf("%s/%d :: FAILED to claim the partition: %s\n", topic, partition, err)
				cg.errors <- &sarama.ConsumerError{
					Topic:     topic,
					Partition: partition,
					Err:       err,
				}
				return
			}
		}
	}

	defer func() {
		err := cg.instance.ReleasePartition(topic, partition)
		if err != nil {
			cg.Logf("%s/%d :: FAILED to release partition: %s\n", topic, partition, err)
			cg.errors <- &sarama.ConsumerError{
				Topic:     topic,
				Partition: partition,
				Err:       err,
			}
		}
	}()

	nextOffset, err := cg.offsetManager.InitializePartition(topic, partition)
	if err != nil {
		cg.Logf("%s/%d :: FAILED to determine initial offset: %s\n", topic, partition, err)
		return
	}

	if nextOffset >= 0 {
		cg.Logf("%s/%d :: Partition consumer starting at offset %d.\n", topic, partition, nextOffset)
	} else {
		nextOffset = cg.config.Offsets.Initial
		if nextOffset == sarama.OffsetOldest {
			cg.Logf("%s/%d :: Partition consumer starting at the oldest available offset.\n", topic, partition)
		} else if nextOffset == sarama.OffsetNewest {
			cg.Logf("%s/%d :: Partition consumer listening for new messages only.\n", topic, partition)
		}
	}

	consumer, err := cg.consumePartition(topic, partition, nextOffset)

	if err != nil {
		cg.Logf("%s/%d :: FAILED to start partition consumer: %s\n", topic, partition, err)
		return
	}

	defer consumer.Close()

	err = nil
	var lastOffset int64 = -1 // aka unknown
partitionConsumerLoop:
	for {
		select {
		case <-stopper:
			break partitionConsumerLoop

		case err := <-consumer.Errors():
			if err == nil {
				cg.Logf("%s/%d :: Consumer encountered an invalid state: re-establishing consumption of partition.\n", topic, partition)

				// Errors encountered (if any) are logged in the consumerPartition function
				var cErr error
				consumer, cErr = cg.consumePartition(topic, partition, lastOffset)
				if cErr != nil {
					break partitionConsumerLoop
				}
				continue partitionConsumerLoop
			}

			for {
				select {
				case errors <- err:
					continue partitionConsumerLoop

				case <-stopper:
					break partitionConsumerLoop
				}
			}

		case message := <-consumer.Messages():
			if message == nil {
				cg.Logf("%s/%d :: Consumer encountered an invalid state: re-establishing consumption of partition.\n", topic, partition)

				// Errors encountered (if any) are logged in the consumerPartition function
				var cErr error
				consumer, cErr = cg.consumePartition(topic, partition, lastOffset)
				if cErr != nil {
					break partitionConsumerLoop
				}
				continue partitionConsumerLoop

			}

			for {
				select {
				case <-stopper:
					break partitionConsumerLoop

				case messages <- message:
					lastOffset = message.Offset
					continue partitionConsumerLoop
				}
			}
		}
	}

	cg.Logf("%s/%d :: Stopping partition consumer at offset %d\n", topic, partition, lastOffset)
	if err := cg.offsetManager.FinalizePartition(topic, partition, lastOffset, cg.config.Offsets.ProcessingTimeout); err != nil {
		cg.Logf("%s/%d :: %s\n", topic, partition, err)
	}
}
1059669
func New(brokers []string, topic, group string) *KafkaSource {
	return &KafkaSource{
		Brokers:        brokers,
		Topic:          topic,
		Group:          group,
		TimeoutSeconds: 16,

		prefix: topic,
	}
}
61557
func newpcpMetricDesc(n string, t MetricType, s MetricSemantics, u MetricUnit, desc ...string) (*pcpMetricDesc, error) {
	if n == "" {
		return nil, errors.New("Metric name cannot be empty")
	}

	if len(n) > StringLength {
		return nil, errors.New("metric name is too long")
	}

	if len(desc) > 2 {
		return nil, errors.New("only 2 optional strings allowed, short and long descriptions")
	}

	shortdesc, longdesc := "", ""

	if len(desc) > 0 {
		shortdesc = desc[0]
	}

	if len(desc) > 1 {
		longdesc = desc[1]
	}

	return &pcpMetricDesc{
		hash(n, PCPMetricItemBitLength),
		n, t, s, u,
		shortdesc, longdesc,
	}, nil
}
612177
func New(addr string) *Metrics {
	met := &Metrics{
		Addr:    addr,
		Reg:     prometheus.NewRegistry(),
		zoneMap: make(map[string]struct{}),
	}
	// Add the default collectors
	met.MustRegister(prometheus.NewGoCollector())
	met.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))

	// Add all of our collectors
	met.MustRegister(buildInfo)
	met.MustRegister(vars.Panic)
	met.MustRegister(vars.RequestCount)
	met.MustRegister(vars.RequestDuration)
	met.MustRegister(vars.RequestSize)
	met.MustRegister(vars.RequestDo)
	met.MustRegister(vars.RequestType)
	met.MustRegister(vars.ResponseSize)
	met.MustRegister(vars.ResponseRcode)
	met.MustRegister(vars.PluginEnabled)

	return met
}
749279
func NewConsumedToken(tkn *lexer.Token) ConsumedElement {
	return ConsumedElement{
		isSymbol: false,
		token:    tkn,
	}
}
663077
func NewPressureStatsCollector() (Collector, error) {
	fs, err := procfs.NewFS(*procPath)
	if err != nil {
		return nil, fmt.Errorf("failed to open procfs: %v", err)
	}

	return &pressureStatsCollector{
		cpu: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "pressure", "cpu_waiting_seconds_total"),
			"Total time in seconds that processes have waited for CPU time",
			nil, nil,
		),
		io: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "pressure", "io_waiting_seconds_total"),
			"Total time in seconds that processes have waited due to IO congestion",
			nil, nil,
		),
		ioFull: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "pressure", "io_stalled_seconds_total"),
			"Total time in seconds no process could make progress due to IO congestion",
			nil, nil,
		),
		mem: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "pressure", "memory_waiting_seconds_total"),
			"Total time in seconds that processes have waited for memory",
			nil, nil,
		),
		memFull: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "pressure", "memory_stalled_seconds_total"),
			"Total time in seconds no process could make progress due to memory congestion",
			nil, nil,
		),
		fs: fs,
	}, nil
}
643117
func NewKadParams() *KadParams {
	return &KadParams{
		MaxProxDisplay:    16,
		NeighbourhoodSize: 2,
		MinBinSize:        2,
		MaxBinSize:        4,
		RetryInterval:     4200000000, // 4.2 sec
		MaxRetries:        42,
		RetryExponent:     2,
	}
}
884666
func (a *Api) podContainerMetrics(request *restful.Request, response *restful.Response) {
	a.processMetricRequest(
		core.PodContainerKey(request.PathParameter("namespace-name"),
			request.PathParameter("pod-name"),
			request.PathParameter("container-name"),
		),
		request, response)
}
108119
func KafkaProducer(p sarama.AsyncProducer) KafkaOption {
	return func(c *KafkaCollector) { c.producer = p }
}
177591
func (pc *KafkaPartitionConsumer) Commit(offset int64) error {
	pc.metrics.NumOffsetCommits(func(numOffsetCommits metrics.Counter) {
		numOffsetCommits.Inc(1)
	})
	err := pc.client.CommitOffset(pc.config.Group, pc.topic, pc.partition, offset)
	if err != nil {
		pc.metrics.NumFailedOffsetCommits(func(numFetchedMessages metrics.Counter) {
			numFetchedMessages.Inc(1)
		})
	}

	return err
}
764698
func NewConsumer(client Client, config *ConsumerConfig, strategy Strategy) Consumer {
	var metrics ConsumerMetrics = noOpConsumerMetrics
	if config.EnableMetrics {
		metrics = NewKafkaConsumerMetrics(config.Group, config.ConsumerID)
	}

	return &KafkaConsumer{
		config:                   config,
		client:                   client,
		strategy:                 strategy,
		metrics:                  metrics,
		partitionConsumers:       make(map[string]map[int32]PartitionConsumer),
		partitionConsumerFactory: NewPartitionConsumer,
		stopped:                  make(chan struct{}),
	}
}
1245415
func (a *AdminClient) CreatePartitions(ctx context.Context, partitions []PartitionsSpecification, options ...CreatePartitionsAdminOption) (result []TopicResult, err error) {
	cParts := make([]*C.rd_kafka_NewPartitions_t, len(partitions))

	cErrstrSize := C.size_t(512)
	cErrstr := (*C.char)(C.malloc(cErrstrSize))
	defer C.free(unsafe.Pointer(cErrstr))

	// Convert Go PartitionsSpecification to C NewPartitions
	for i, part := range partitions {
		cParts[i] = C.rd_kafka_NewPartitions_new(C.CString(part.Topic), C.size_t(part.IncreaseTo), cErrstr, cErrstrSize)
		if cParts[i] == nil {
			return nil, newErrorFromString(ErrInvalidArg,
				fmt.Sprintf("Topic %s: %s", part.Topic, C.GoString(cErrstr)))
		}

		defer C.rd_kafka_NewPartitions_destroy(cParts[i])

		for pidx, replicas := range part.ReplicaAssignment {
			cReplicas := make([]C.int32_t, len(replicas))
			for ri, replica := range replicas {
				cReplicas[ri] = C.int32_t(replica)
			}
			cErr := C.rd_kafka_NewPartitions_set_replica_assignment(
				cParts[i], C.int32_t(pidx),
				(*C.int32_t)(&cReplicas[0]), C.size_t(len(cReplicas)),
				cErrstr, cErrstrSize)
			if cErr != 0 {
				return nil, newCErrorFromString(cErr,
					fmt.Sprintf("Failed to set replica assignment for topic %s new partition index %d: %s", part.Topic, pidx, C.GoString(cErrstr)))
			}
		}

	}

	// Convert Go AdminOptions (if any) to C AdminOptions
	genericOptions := make([]AdminOption, len(options))
	for i := range options {
		genericOptions[i] = options[i]
	}
	cOptions, err := adminOptionsSetup(a.handle, C.RD_KAFKA_ADMIN_OP_CREATEPARTITIONS, genericOptions)
	if err != nil {
		return nil, err
	}
	defer C.rd_kafka_AdminOptions_destroy(cOptions)

	// Create temporary queue for async operation
	cQueue := C.rd_kafka_queue_new(a.handle.rk)
	defer C.rd_kafka_queue_destroy(cQueue)

	// Asynchronous call
	C.rd_kafka_CreatePartitions(
		a.handle.rk,
		(**C.rd_kafka_NewPartitions_t)(&cParts[0]),
		C.size_t(len(cParts)),
		cOptions,
		cQueue)

	// Wait for result, error or context timeout
	rkev, err := a.waitResult(ctx, cQueue, C.RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT)
	if err != nil {
		return nil, err
	}
	defer C.rd_kafka_event_destroy(rkev)

	cRes := C.rd_kafka_event_CreatePartitions_result(rkev)

	// Convert result from C to Go
	var cCnt C.size_t
	cTopicRes := C.rd_kafka_CreatePartitions_result_topics(cRes, &cCnt)

	return a.cToTopicResults(cTopicRes, cCnt)
}
746978
func newFileMetrics(labels prometheus.Labels) *fileMetrics {
	var names []string
	for k := range labels {
		names = append(names, k)
	}
	names = append(names, "level")
	sort.Strings(names)

	return &fileMetrics{
		DiskSize: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Namespace: namespace,
			Subsystem: fileStoreSubsystem,
			Name:      "disk_bytes",
			Help:      "Number of bytes TSM files using on disk.",
		}, names),
		Files: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Namespace: namespace,
			Subsystem: fileStoreSubsystem,
			Name:      "total",
			Help:      "Number of files.",
		}, names),
	}
}