[MediaWiki-commits] [Gerrit] Set up kafka processor and kafka consumer for client side ev... - change (operations/puppet)

[Ottomata (Code Review)]

Ottomata has uploaded a new change for review.

  https://gerrit.wikimedia.org/r/197418

Change subject: Set up kafka processor and kafka consumer for client side 
events from /beacon/event.gif via varnishkafka
......................................................................

Set up kafka processor and kafka consumer for client side events from 
/beacon/event.gif via varnishkafka

Change-Id: I2f9a1655894b99f9a7e35c5c9626a769ceddce9a
---
M manifests/role/eventlogging.pp
1 file changed, 28 insertions(+), 22 deletions(-)


  git pull ssh://gerrit.wikimedia.org:29418/operations/puppet 
refs/changes/18/197418/1

diff --git a/manifests/role/eventlogging.pp b/manifests/role/eventlogging.pp
index 32713fb..d307dc9 100644
--- a/manifests/role/eventlogging.pp
+++ b/manifests/role/eventlogging.pp
@@ -18,6 +18,8 @@
 #
 class role::eventlogging {
     include eventlogging
+    # Infer Kafka cluster configuration from this class
+    include role::analytics::kafka::config
 
     if hiera('has_ganglia', true) {
         include role::eventlogging::monitoring
@@ -67,12 +69,33 @@
         output => 'tcp://*:8522',
     }
 
-    # Parsed and validated client-side (Varnish-generated) and
+
+
+    # Client-side events can also be generated by requests to
+    # //bits.wikimedia.org/beacon.event.gif?<event payload>.
+    # A varnishkafka instance on the bits caches sends these
+    # requests to kafka under the 'eventlogging' topic.
+    # Eventlogging can consume from this topic using kafka://
+    # URIs.
+    $kafka_brokers inline_template('<%= 
scope.lookupvar("role::analytics::kafka::config::brokers_array").join(",") %>')
+    $kafka_eventlogging_input_topic = 'eventlogging'
+    $kafka_input_uri = 
"kafka://${kafka_brokers}?topic=${kafka_eventlogging_input_topic}"
+
+    # This processor consumes client side events from Kafka and sends them
+    # to zmq port 8523.
+    eventlogging::service::processor { 'client-side events kafka':
+        format => '%q %{recvFrom}s %{seqId}d %t %h %{userAgent}i',
+        input  => 
"${kafka_input_uri}&group_id=eventlogging-client-side-processor",
+        output => 'tcp://*:8523',
+    }
+
+
+    # Parsed and validated client-side (varnishncsa and varnishkafka 
generated) and
     # server-side (MediaWiki-generated) events are multiplexed into a
     # single output stream, published on TCP port 8600.
 
     eventlogging::service::multiplexer { 'all events':
-        inputs => [ "tcp://${processor}:8521", "tcp://${processor}:8522" ],
+        inputs => [ "tcp://${processor}:8521", "tcp://${processor}:8522", 
"tcp://${processor}:8523" ],
         output => 'tcp://*:8600',
     }
 
@@ -109,6 +132,9 @@
         'client-side-events.log':
             input  => "tcp://${processor}:8422?raw=1",
             output => "file://${log_dir}/client-side-events.log";
+        'client-side-events-kafka.log':
+            input  => 
"${kafka_input_uri}&group_id=eventlogging-client-side-raw-consumer"
+            output => "file://${log_dir}/client-side-events-kafka.log";
         'all-events.log':
             input  => "tcp://${processor}:8600",
             output => "file://${log_dir}/all-events.log";
@@ -129,26 +155,6 @@
             require     => File[$log_dir],
             hosts_allow => $backup_destinations,
         }
-    }
-
-
-    ## Kafka / Hadoop
-
-    include role::analytics::kafka::config
-
-    $kafka_brokers = inline_template('<%= 
scope.lookupvar("role::analytics::kafka::config::brokers_array").join(",") %>')
-    $kafka_cluster = $::role::analytics::kafka::config::kafka_cluster_name
-
-    $kafka_topic_base_name = 'eventlogging'
-    $kafka_topic_version = 0
-    $kafka_topic = sprintf('%s-%02d', $kafka_topic_base_name, 
$kafka_topic_version)
-
-    eventlogging::service::consumer { 'kafka':
-        input  => "tcp://${processor}:8600",
-        output => 
"kafka://${kafka_cluster}?brokers=${kafka_brokers}&topic=${kafka_topic}",
-        # We are not currently using this data in Kafka or Hadoop, and Kafka 
is about
-        # to undergo some production failover testing.  Disabling this for now.
-        ensure => 'absent',
     }
 }
 

-- 
To view, visit https://gerrit.wikimedia.org/r/197418
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: I2f9a1655894b99f9a7e35c5c9626a769ceddce9a
Gerrit-PatchSet: 1
Gerrit-Project: operations/puppet
Gerrit-Branch: production
Gerrit-Owner: Ottomata <o...@wikimedia.org>

_______________________________________________
MediaWiki-commits mailing list
MediaWiki-commits@lists.wikimedia.org
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits