Skip to content
Merged
Show file tree
Hide file tree
Changes from 13 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions metricbeat/docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ services:
beat:
build: ${PWD}/.
environment:
- BEAT_STRICT_PERMS=false
- TEST_ENVIRONMENT=false
working_dir: /go/src/github.com/elastic/beats/metricbeat
volumes:
Expand Down
105 changes: 102 additions & 3 deletions metricbeat/docs/fields.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -15436,6 +15436,24 @@ type: float

--

*`kafka.broker.request.produce.failed`*::
+
--
The number of failed produce requests

type: float

--

*`kafka.broker.request.fetch.failed`*::
+
--
The number of client fetch request failures

type: float

--

*`kafka.broker.replication.leader_elections`*::
+
--
Expand Down Expand Up @@ -15502,7 +15520,7 @@ type: float
*`kafka.broker.topic.net.bytes_in`*::
+
--
The incoming byte rate
The incoming byte rate per topic

type: float

Expand All @@ -15511,7 +15529,7 @@ type: float
*`kafka.broker.topic.net.bytes_out`*::
+
--
The outgoing byte rate
The outgoing byte rate per topic

type: float

Expand All @@ -15520,7 +15538,7 @@ type: float
*`kafka.broker.topic.net.bytes_rejected`*::
+
--
The rejected byte rate
The rejected byte rate per topic

type: float

Expand All @@ -15529,6 +15547,42 @@ type: float
*`kafka.broker.topic.messages_in`*::
+
--
The incoming message rate per topic

type: float

--

*`kafka.broker.net.bytes_in`*::
+
--
The incoming byte rate

type: float

--

*`kafka.broker.net.bytes_out`*::
+
--
The outgoing byte rate

type: float

--

*`kafka.broker.net.bytes_rejected`*::
+
--
The rejected byte rate

type: float

--

*`kafka.broker.messages_in`*::
+
--
The incoming message rate

type: float
Expand Down Expand Up @@ -15586,6 +15640,42 @@ type: float

--

*`kafka.consumer.max_lag`*::
+
--
The maximum consumer lag

type: float

--

*`kafka.consumer.zookeeper_commits`*::
+
--
The rate of offset commits to ZooKeeper

type: float

--

*`kafka.consumer.kafka_commits`*::
+
--
The rate of offset commits to Kafka

type: float

--

*`kafka.consumer.messages_in`*::
+
--
The rate of consumer message consumption

type: float

--

[float]
=== consumergroup

Expand Down Expand Up @@ -16013,6 +16103,15 @@ type: float

--

*`kafka.producer.message_rate`*::
+
--
The producer message rate

type: float

--

[[exported-fields-kibana]]
== Kibana fields

Expand Down
24 changes: 21 additions & 3 deletions metricbeat/module/kafka/broker/_meta/fields.yml
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,12 @@
- name: request.fetch.failed_per_second
description: The rate of client fetch request failures per second
type: float
- name: request.produce.failed
description: The number of failed produce requests
type: float
- name: request.fetch.failed
description: The number of client fetch request failures
type: float
- name: replication.leader_elections
description: The leader election rate
type: float
Expand All @@ -37,14 +43,26 @@
description: The log flush rate
type: float
- name: topic.net.bytes_in
description: The incoming byte rate
description: The incoming byte rate per topic
type: float
- name: topic.net.bytes_out
description: The outgoing byte rate
description: The outgoing byte rate per topic
type: float
- name: topic.net.bytes_rejected
description: The rejected byte rate
description: The rejected byte rate per topic
type: float
- name: topic.messages_in
description: The incoming message rate per topic
type: float
- name: net.bytes_in

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Following field conventions this should probably be:

Suggested change
- name: net.bytes_in
- name: net.in.bytes_per_sec

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@sorantis is this ok from product perspective or we should strictly follow the naming provided in the initial issue?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

there's no strict requirement on the naming. The field names were taken from the ER attached to the original issue. Do we have the mentioned naming convention somewhere?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

description: The incoming byte rate
type: float
- name: net.bytes_out

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And:

Suggested change
- name: net.bytes_out
- name: net.out.bytes_per_sec

description: The outgoing byte rate
type: float
- name: net.bytes_rejected

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And:

Suggested change
- name: net.bytes_rejected
- name: net.rejected.bytes_per_sec

description: The rejected byte rate
type: float
- name: messages_in
description: The incoming message rate
type: float
66 changes: 66 additions & 0 deletions metricbeat/module/kafka/broker/broker_integration_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. 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
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

// +build integration

package broker

import (
"testing"
"time"

"github.com/elastic/beats/libbeat/tests/compose"
mbtest "github.com/elastic/beats/metricbeat/mb/testing"
// Register input module and metricset
_ "github.com/elastic/beats/metricbeat/module/jolokia"
_ "github.com/elastic/beats/metricbeat/module/jolokia/jmx"
)

func TestData(t *testing.T) {
service := compose.EnsureUp(t, "kafka",
compose.UpWithTimeout(600*time.Second),
compose.UpWithAdvertisedHostEnvFileForPort(9092),
)

ms := mbtest.NewReportingMetricSetV2Error(t, getConfig(service.HostForPort(8779)))
err := mbtest.WriteEventsReporterV2Error(ms, t, "")
if err != nil {
t.Fatal("write", err)
}
Comment thread
ChrsMark marked this conversation as resolved.
Outdated
}

func TestFetch(t *testing.T) {
service := compose.EnsureUp(t, "kafka",
compose.UpWithTimeout(600*time.Second),
compose.UpWithAdvertisedHostEnvFileForPort(9092),
)
reporter := &mbtest.CapturingReporterV2{}

metricSet := mbtest.NewReportingMetricSetV2Error(t, getConfig(service.HostForPort(8779)))
Comment thread
ChrsMark marked this conversation as resolved.
Outdated
metricSet.Fetch(reporter)

e := mbtest.StandardizeEvent(metricSet, reporter.GetEvents()[0])
t.Logf("%s/%s event: %+v", metricSet.Module().Name(), metricSet.Name(), e.Fields.StringToPrint())
}

func getConfig(host string) map[string]interface{} {
return map[string]interface{}{
"module": "kafka",
"metricsets": []string{"broker"},
"hosts": []string{host},
}
}
33 changes: 33 additions & 0 deletions metricbeat/module/kafka/broker/broker_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. 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
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

package broker

import (
"os"

"github.com/elastic/beats/metricbeat/mb"
// Register input module and metricset
_ "github.com/elastic/beats/metricbeat/module/jolokia"
_ "github.com/elastic/beats/metricbeat/module/jolokia/jmx"
)

func init() {
// To be moved to some kind of helper
os.Setenv("BEAT_STRICT_PERMS", "false")
mb.Registry.SetSecondarySource(mb.NewLightModulesSource("../../../module"))
}
34 changes: 30 additions & 4 deletions metricbeat/module/kafka/broker/manifest.yml
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,14 @@ input:
attributes:
- attr: MeanRate
field: request.fetch.failed_per_second
- mbean: 'kafka.server:name=FailedProduceRequestsPerSec,type=BrokerTopicMetrics'
attributes:
- attr: MeanRate
field: request.produce.failed
- mbean: 'kafka.server:name=FailedFetchRequestsPerSec,type=BrokerTopicMetrics'
attributes:
- attr: MeanRate
field: request.fetch.failed
- mbean: 'kafka.controller:name=LeaderElectionRateAndTimeMs,type=ControllerStats'
attributes:
- attr: MeanRate
Expand Down Expand Up @@ -47,19 +55,37 @@ input:
attributes:
- attr: MeanRate
field: log.flush_rate
- mbean: 'kafka.server:name=BytesRejectedPerSec,type=BrokerTopicMetrics'
- mbean: 'kafka.server:name=BytesRejectedPerSec,topic=*,type=BrokerTopicMetrics'
attributes:
- attr: MeanRate
field: topic.net.bytes_rejected
- mbean: 'kafka.server:name=BytesInPerSec,type=BrokerTopicMetrics'
- mbean: 'kafka.server:name=BytesInPerSec,topic=*,type=BrokerTopicMetrics,topic=*'
attributes:
- attr: MeanRate
field: topic.net.bytes_in
- mbean: 'kafka.server:name=BytesOutPerSec,type=BrokerTopicMetrics'
- mbean: 'kafka.server:name=BytesOutPerSec,topic=*,type=BrokerTopicMetrics,topic=*'
attributes:
- attr: MeanRate
field: topic.net.bytes_out
- mbean: 'kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec'
- mbean: 'kafka.server:type=BrokerTopicMetrics,topic=*,name=MessagesInPerSec,topic=*'
attributes:
- attr: MeanRate
field: topic.messages_in
- mbean: 'kafka.server:name=BytesRejectedPerSec,type=BrokerTopicMetrics'
attributes:
- attr: MeanRate
field: net.bytes_rejected
- mbean: 'kafka.server:name=BytesInPerSec,type=BrokerTopicMetrics'
attributes:
- attr: MeanRate
field: net.bytes_in
- mbean: 'kafka.server:name=BytesOutPerSec,type=BrokerTopicMetrics'
attributes:
- attr: MeanRate
field: net.bytes_out
- mbean: 'kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec'
attributes:
- attr: MeanRate
field: messages_in


12 changes: 12 additions & 0 deletions metricbeat/module/kafka/consumer/_meta/fields.yml
Original file line number Diff line number Diff line change
Expand Up @@ -18,3 +18,15 @@
- name: bytes_in
description: The rate of bytes coming in to the consumer
type: float
- name: max_lag
description: The maximum consumer lag
type: float
- name: zookeeper_commits
description: The rate of offset commits to ZooKeeper
type: float
- name: kafka_commits
description: The rate of offset commits to Kafka
type: float
- name: messages_in
description: The rate of consumer message consumption
type: float
Loading