JIRA: SAMZA-1733 Specify MetricsSnapshotStream partitionKey in config #542
Conversation
rayman7718
commented
May 31, 2018
rayman7718
commented
- Adding config to specify MetricsSnapshotStream partitionKey, defaults to hostname (to maintain backword compat)
… to hostname (to maintain backword compat)
| * @param systemStream The systemStream for the MetricsSnapshotStream | ||
| * @return the metric key name (if specified), else None | ||
| */ | ||
| def getMetricsReporterStreamPartitionKeyName(systemStream: SystemStream): Option[String] = getOption(MetricsConfig.METRICS_SNAPSHOT_REPORTER_PARTITION_KEY format(systemStream.getSystem, systemStream.getStream)) |
There was a problem hiding this comment.
Should expose these configs independent of the systemStream. One alternative:
reporters.snapshot.key whose values can be in the set: {host, jobId, jobName}
…an by systemStream
| val METRICS_SNAPSHOT_REPORTER_INTERVAL = "metrics.reporter.%s.interval" | ||
| val METRICS_TIMER_ENABLED = "metrics.timer.enabled" | ||
| val METRICS_SNAPSHOT_REPORTER_PARTITION_KEY = "metrics.reporter.%s.partitionkey" | ||
| val METRICS_SNAPSHOT_REPORTER_PARTITION_KEY_JOBNAME = "jobname" |
There was a problem hiding this comment.
s/partition_key_jobname/partition_key_default
| * @param jobName the current job name | ||
| * @param hostname the current host name | ||
| */ | ||
| def getMetricsReporterStreamPartitionKeyValue(keyName: Option[String], jobName: String, hostname: String) = keyName match { |
There was a problem hiding this comment.
prefer moving this method out of the MetricReporterConfig and instead adding it to SnapshotReporter instead.
…moving getPartitionValue method to MetricsSnapshotReporter
| val METRICS_TIMER_ENABLED= "metrics.timer.enabled" | ||
| val METRICS_SNAPSHOT_REPORTER_INTERVAL = "metrics.reporter.%s.interval" | ||
| val METRICS_TIMER_ENABLED = "metrics.timer.enabled" | ||
| val METRICS_SNAPSHOT_REPORTER_PARTITION_KEY = "metrics.reporter.%s.partitionkey" |
There was a problem hiding this comment.
Just curious, shouldn't we have a default value for this configuration variable?
| import java.util.HashMap | ||
| import java.util.Map | ||
| import java.util.concurrent.Executors | ||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| import scala.collection.JavaConverters._ | ||
|
|
||
| object MetricsSnapshotReporter { | ||
| val METRICS_SNAPSHOT_REPORTER_PARTITION_KEY_DEFAULT = "jobname" |
There was a problem hiding this comment.
We use a pattern where the default value is set in config class's getXXXConfig() methods, instead of the static defaults in the implementation class.
rayman7718
changed the title
rayman7718
changed the title
| def getMetricsReporterStreamPartitionKeyValue(keyName: String, jobName: String, hostname: String) = keyName match { | ||
| case "jobname" => jobName | ||
| case "hostname" => hostname | ||
| case _ => keyName |
There was a problem hiding this comment.
Is this keyName or keyValue? The javadoc says "else returns the actual value specified", but the input argument only have key name. In addition, if "jobname" and "hostname" are special key names, I would prefer to define those as constant in MetricsConfig.
