github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 06:37:19 -04:00
Code Issues Projects Releases Packages Wiki Activity

avoid race condition when initializing events and pipelines metrics (#9950)

Browse source
This commit is contained in:
Colin Surprenant 2018-09-04 13:37:40 -04:00 • committed by GitHub
parent 796eb3e4b3
commit 5a49102b89
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 39 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

22
logstash-core/src/main/java/org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java
View file

@ -106,16 +106,18 @@ public abstract class AbstractOutputDelegatorExt extends RubyObject {
this.metric = metric;
final ThreadContext context = RubyUtil.RUBY.getCurrentContext();
this.id = RubyString.newString(context.runtime, id);
namespacedMetric = metric.namespace(context, context.runtime.newSymbol(id));
metricEvents = namespacedMetric.namespace(context, MetricKeys.EVENTS_KEY);
namespacedMetric.gauge(
context, MetricKeys.NAME_KEY, configName(context)
);
eventMetricOut = LongCounter.fromRubyBase(metricEvents, MetricKeys.OUT_KEY);
eventMetricIn = LongCounter.fromRubyBase(metricEvents, MetricKeys.IN_KEY);
eventMetricTime = LongCounter.fromRubyBase(
metricEvents, MetricKeys.DURATION_IN_MILLIS_KEY
);
synchronized (metric) {
namespacedMetric = metric.namespace(context, context.runtime.newSymbol(id));
metricEvents = namespacedMetric.namespace(context, MetricKeys.EVENTS_KEY);
namespacedMetric.gauge(
context, MetricKeys.NAME_KEY, configName(context)
);
eventMetricOut = LongCounter.fromRubyBase(metricEvents, MetricKeys.OUT_KEY);
eventMetricIn = LongCounter.fromRubyBase(metricEvents, MetricKeys.IN_KEY);
eventMetricTime = LongCounter.fromRubyBase(
metricEvents, MetricKeys.DURATION_IN_MILLIS_KEY
);
}
}
protected abstract IRubyObject getConfigName(ThreadContext context);

17
logstash-core/src/main/java/org/logstash/config/ir/compiler/FilterDelegatorExt.java
View file

@ -53,15 +53,14 @@ public final class FilterDelegatorExt extends RubyObject {
this.filter = filter;
filterClass = filter.getSingletonClass().getRealClass();
filterMethod = filterClass.searchMethod(FILTER_METHOD_NAME);
final AbstractNamespacedMetricExt namespacedMetric =
(AbstractNamespacedMetricExt) filter.callMethod(context, "metric");
metricEvents = namespacedMetric.namespace(context, MetricKeys.EVENTS_KEY);
eventMetricOut = LongCounter.fromRubyBase(metricEvents, MetricKeys.OUT_KEY);
eventMetricIn = LongCounter.fromRubyBase(metricEvents, MetricKeys.IN_KEY);
eventMetricTime = LongCounter.fromRubyBase(
metricEvents, MetricKeys.DURATION_IN_MILLIS_KEY
);
namespacedMetric.gauge(context, MetricKeys.NAME_KEY, configName(context));
final AbstractNamespacedMetricExt namespacedMetric = (AbstractNamespacedMetricExt) filter.callMethod(context, "metric");
synchronized(namespacedMetric.getMetric()) {
metricEvents = namespacedMetric.namespace(context, MetricKeys.EVENTS_KEY);
eventMetricOut = LongCounter.fromRubyBase(metricEvents, MetricKeys.OUT_KEY);
eventMetricIn = LongCounter.fromRubyBase(metricEvents, MetricKeys.IN_KEY);
eventMetricTime = LongCounter.fromRubyBase(metricEvents, MetricKeys.DURATION_IN_MILLIS_KEY);
namespacedMetric.gauge(context, MetricKeys.NAME_KEY, configName(context));
}
flushes = filter.respondsTo("flush");
return this;
}

17
logstash-core/src/main/java/org/logstash/execution/QueueReadClientBase.java
View file

@ -52,19 +52,22 @@ public abstract class QueueReadClientBase extends RubyObject implements QueueRea
@JRubyMethod(name = "set_events_metric")
public IRubyObject setEventsMetric(final IRubyObject metric) {
final AbstractNamespacedMetricExt namespacedMetric = (AbstractNamespacedMetricExt) metric;
eventMetricOut = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.OUT_KEY);
eventMetricFiltered = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.FILTERED_KEY);
eventMetricTime = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.DURATION_IN_MILLIS_KEY);
synchronized(namespacedMetric.getMetric()) {
eventMetricOut = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.OUT_KEY);
eventMetricFiltered = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.FILTERED_KEY);
eventMetricTime = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.DURATION_IN_MILLIS_KEY);
}
return this;
}
@JRubyMethod(name = "set_pipeline_metric")
public IRubyObject setPipelineMetric(final IRubyObject metric) {
final AbstractNamespacedMetricExt namespacedMetric = (AbstractNamespacedMetricExt) metric;
pipelineMetricOut = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.OUT_KEY);
pipelineMetricFiltered = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.FILTERED_KEY);
pipelineMetricTime =
LongCounter.fromRubyBase(namespacedMetric, MetricKeys.DURATION_IN_MILLIS_KEY);
synchronized(namespacedMetric.getMetric()) {
pipelineMetricOut = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.OUT_KEY);
pipelineMetricFiltered = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.FILTERED_KEY);
pipelineMetricTime = LongCounter.fromRubyBase(namespacedMetric, MetricKeys.DURATION_IN_MILLIS_KEY);
}
return this;
}

2
logstash-core/src/main/java/org/logstash/instrument/metrics/AbstractNamespacedMetricExt.java
View file

@ -68,4 +68,6 @@ public abstract class AbstractNamespacedMetricExt extends AbstractMetricExt {
protected abstract IRubyObject doIncrement(ThreadContext context, IRubyObject[] args);
protected abstract IRubyObject doDecrement(ThreadContext context, IRubyObject[] args);
public abstract AbstractMetricExt getMetric();
}

3
logstash-core/src/main/java/org/logstash/instrument/metrics/NamespacedMetricExt.java
View file

@ -103,4 +103,7 @@ public final class NamespacedMetricExt extends AbstractNamespacedMetricExt {
name instanceof RubyArray ? name : RubyArray.newArray(context.runtime, name)
));
}
@Override
public AbstractMetricExt getMetric() { return this.metric; }
}

5
logstash-core/src/main/java/org/logstash/instrument/metrics/NullNamespacedMetricExt.java
View file

@ -37,7 +37,7 @@ public final class NullNamespacedMetricExt extends AbstractNamespacedMetricExt {
@JRubyMethod(optional = 2)
public NullNamespacedMetricExt initialize(final ThreadContext context,
final IRubyObject[] args) {
this.metric = args.length > 0 && !args[0].isNil() ? (NullMetricExt) args[0] : null;
this.metric = args.length > 0 && !args[0].isNil() ? (NullMetricExt) args[0] : new NullMetricExt(context.runtime, metaClass);
final IRubyObject namespaceName = args.length == 2 ? args[1] : NULL;
if (namespaceName instanceof RubyArray) {
this.namespaceName = (RubyArray) namespaceName;
@ -99,6 +99,9 @@ public final class NullNamespacedMetricExt extends AbstractNamespacedMetricExt {
));
}
@Override
public AbstractMetricExt getMetric() { return this.metric; }
@JRubyClass(name = "NullCounter")
public static final class NullCounter extends RubyObject {