blob: 33d288f7cc0ebc13988e4c82204f6b81cbe6f487 [file] [log] [blame]
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
package org.apache.samza.container
import java.util
import org.apache.samza.diagnostics.DiagnosticsExceptionEvent
import org.apache.samza.metrics.{Gauge, ReadableMetricsRegistry, MetricsRegistryMap, MetricsHelper}
class SamzaContainerMetrics(
val source: String = "unknown",
val registry: ReadableMetricsRegistry = new MetricsRegistryMap,
val prefix: String = "") extends MetricsHelper {
val commits = newCounter("commit-calls")
val windows = newCounter("window-calls")
val timers = newCounter("timer-calls")
val processes = newCounter("process-calls")
val envelopes = newCounter("process-envelopes")
val nullEnvelopes = newCounter("process-null-envelopes")
val chooseNs = newTimer("choose-ns")
val windowNs = newTimer("window-ns")
val timerNs = newTimer("timer-ns")
val processNs = newTimer("process-ns")
val commitNs = newTimer("commit-ns")
val blockNs = newTimer("block-ns")
val containerStartupTime = newGauge("container-startup-time", 0L)
val utilization = newGauge("event-loop-utilization", 0.0F)
val diskUsageBytes = newGauge("disk-usage-bytes", 0L)
val diskQuotaBytes = newGauge("disk-quota-bytes", Long.MaxValue)
val executorWorkFactor = newGauge("executor-work-factor", 1.0)
val physicalMemoryMb = newGauge("physical-memory-mb", 0.0F)
val physicalMemoryUtilization = newGauge("physical-memory-utilization", 0.0F)
val containerThreadPoolSize = newGauge("container-thread-pool-size", 0L)
val containerActiveThreads = newGauge("container-active-threads", 0L)
val taskStoreRestorationMetrics: util.Map[TaskName, Gauge[Long]] = new util.HashMap[TaskName, Gauge[Long]]()
def addStoresRestorationGauge(taskName: TaskName) {
taskStoreRestorationMetrics.put(taskName, newGauge("%s-restore-time" format(taskName.toString), -1L))
override def getPrefix: String = prefix