You are browsing a read-only backup copy of Wikitech. The live site can be found at wikitech.wikimedia.org

Event Platform/EventBus: Difference between revisions

From Wikitech-static
Jump to navigation Jump to search
imported>Ottomata
No edit summary
imported>Ottomata
No edit summary
 
(One intermediate revision by the same user not shown)
Line 1: Line 1:
#REDIRECT [[mw:Extension:EventBus]]


{{Note|eventlogging-service-eventbus has be deprecated and decommissioned in favor of various EventGate services.  'EventBus' now refers only to the Mediawiki EventBus extension which is responsible for sending events to [[Event_Platform/EventGate|EventGate]].  This page is left for posterity but should not longer be used as a reference.}}
{{Navigation Event Platform}}
 
If you need to send or collect data between or from different services, applications or features (for example, propagating change events from MediaWiki to a service), you should consider using EventBus.
 
The term 'EventBus' does not refer to one specific technology.  It refers to the WMF's use of various technologies to distribute streams of schema-conforming messages, or ''events''.
 
More concretely, it is a [https://gerrit.wikimedia.org/r/#/admin/projects/mediawiki/event-schemas common repository] of [http://json-schema.org JSON schema], and a RESTful interface on top of [http://kafka.apache.org Kafka] that performs schema validation.  'EventBus service' refers to the RESTful service that accepts and validates events, and produces to Kafka.
==Administration==
See the [[EventBus/Administration]] page for administration tips and documentation.
== Background ==
Over time the Wikimedia architecture has become more and more services-oriented, making propagation of state changes between disparate systems increasingly important.  In light of this, and the lack of a standardized approach, requirements for a general-purpose publish-subscribe event system were drafted<ref>[[phab:T84923|https://phabricator.wikimedia.org/T84923]], Reliable publish / subscribe event bus</ref><ref>[[m:Research:MediaWiki_events:_a_generalized_public_event_datasource|https://meta.wikimedia.org/wiki/Research:MediaWiki_events:_a_generalized_public_event_datasource]], Research:MediaWiki events: a generalized public event datasource</ref>.  The systems currently in use at Wikimedia were evaluated against these requirements, and of those, [[Analytics/EventLogging|EventLogging]] ultimately proved the most interesting.
 
Initially designed for analytics, [[Analytics/EventLogging|EventLogging]] has been in use at Wikimedia for a number of years, collecting data via an [[Mw:Extension:EventLogging|extension]] and a client-side Javascript library.  Received events are validated against schema maintained in an [https://meta.wikimedia.org/w/index.php?title=Special%3AAllPages&from=&to=&namespace=470 on-wiki repository].  A Python framework allows for composition of arbitrary stream topologies using a wide variety of protocol/transport options.  In production at Wikimedia, the pipeline for analytics was initially composed of UDP and ZeroMQ transports, but in the summer of 2015, Kafka support was added, and nearly all EventLogging processes were ported to it.
 
After considerable discussion of requirements and available options, including an [https://phabricator.wikimedia.org/E86 RFC meeting], a decision to use EventLogging with Kafka was made. Refactoring was performed to make the code-base more general (less specific to Mediawiki), and an HTTP REST API for event production was implemented.
 
=== EventLogging Analytics vs EventBus ===
EventLogging Analytics and EventBus are similar systems (the EventBus HTTP service is implemented in the EventLogging codebase!), but are for slightly different purposes.  If you are a team that wants to generate MediaWiki specific analytics data for your own usage (not shared across teams), and for a relatively short term period of time (less than 2 years), the EventLogging Analytics platform should work well for you.  There, schemas are stored and evolved on wiki.  Events are automatically collected and validated and stored in MySQL and Hive.  Events can be sent from client side browsers or apps.  However, this system does not have strong reliability guarantees (as it is analytics data), and the schemas are not designed with maintainability and share-ability in mind.  This leads to duplicate efforts between teams, and frustration as future developers attempt to adapt data designed for a single analytics purpose to another.
 
EventBus schemas are reviewed more closely than EventLogging Analytics, and are of a different non MediaWiki specific format.  As such, they are (more) consistent (and hopefully) more future proof than EventLogging Analytics schemas usually are.  In production, EventBus runs in multiple Data Centers, and has production service level reliability.  If you want production level event data that can be used by multiple teams and services, you'll want to use EventBus.
 
(Ideally, these services would not be as different as they are.  In the (probably distant) future, we'd like to modify EventLogging Analytics so that it looks a little more like EventBus, and possibly display EventBus schemas on wiki.  This work is not prioritized.)
 
== Event Schemas ==
 
In a system where the writers of data may not be the same as the readers, the format of the data is important.  Data is not static, and developers will make changes to formats over time.  Without a predefined and shared schema, a writer may make a format change without notifying a reader and break the reader's code.  A schema acts as a contract between writers and readers, decoupling the two.  When a schema is changed in a backwards compatible way, developers can change their schema and produce new data without breaking a downstream reader's code.<ref>http://www.confluent.io/blog/stream-data-platform-2, Putting Apache Kafka To Use: A Practical Guide to Building a Stream Data Platform (Part 2)</ref>
 
Event schemas are currently shared via the [https://github.com/wikimedia/mediawiki-event-schemas mediawiki-event-schemas repository].  For now, the EventBus HTTP service only supports JSON Schemas (Avro was considered, but was found to be [[wmfblog:2017/01/13/json-hadoop-kafka/|difficult to work with]]).  JSON Schemas do not support any fancy backwards compatible schema evolution on their own.  However, so long as new fields are only ever added when evolving schemas, then changes can be considered to be backwards compatible.
 
''Note: As of this writing (January 2016), there is no official review process for schema, and there is nothing to prevent you from making backward incompatible changes.  In the future, continuous integration work is planned that will help with this.<sup>[[https://en.wikipedia.org/wiki/Wikipedia:Citation_needed citation needed]]</sup>  In the meantime, please get someone from Services and/or Analytics to review any proposed schema changes.''
 
=== Metadata ===
 
All events share in common a sub-object named <code>meta</code>, that encapsulates a number of standard fields.  The <code>meta</code> sub-object should conform to the following schema:
 
<syntaxhighlight lang="yaml">
meta:
type: object
properties:
  topic:
    type: string
    description: the queue topic name this message belongs to
  schema_uri:
    type: string
    description: >
      The URI identifying the jsonschema for this event.  This may be just
      a short uri containing only the name and revision at the end of the
      URI path.  e.g. schema_name/12345 is acceptable.  This field
      is not required.
  uri:
    type: string
    format: uri
    description: the unique URI identifying the event
  request_id:
    type: string
    pattern: '^[a-fA-F0-9]{8}(-[a-fA-F0-9]{4}){3}-[a-fA-F0-9]{12}$'
    description: the unique UUID v1 ID of the event derived from the X-Request-Id header
  id:
    type: string
    pattern: '^[a-fA-F0-9]{8}(-[a-fA-F0-9]{4}){3}-[a-fA-F0-9]{12}$'
    description: the UUID of this event; should match the dt field if
using UUID1
  dt:
    type: string
    format: date-time
    description: the time stamp of the event, in ISO8601 format
  domain:
    type: string
    description: the domain the event pertains to
required:
  - topic
  - uri
  - id
  - dt
  - domain
</syntaxhighlight>
 
''Note: Future iterations may make use of [https://tools.ietf.org/html/rfc6901 JSON Pointers] ([http://spacetelescope.github.io/understanding-json-schema/structuring.html $refs] in JSON schema parlance) to include this meta-data schema, however for now it should be manually included into every schema.''
 
''Note: As a convenience, JSON Schema should be represented using [http://www.yaml.org/ YAML] (as in the example above), and will be transparently converted to JSON by the EventLogging service.''
 
== RESTful Service (Production) ==
<tt>[https://github.com/wikimedia/eventlogging#eventlogging-service eventlogging-service]</tt> is a generic daemon that receives events via HTTP and produces them to EventLogging outputs.  <tt>eventlogging-service-eventbus</tt> is an instance of <tt>eventlogging-service</tt> specifically for receiving events and producing them to Kafka topics using a strict [https://github.com/wikimedia/mediawiki-event-schemas/blob/master/config/eventbus-topics.yaml topic -> schema mapping].  That is, it has a topic configuration that specifies what type of schema may be produced to a particular Kafka topic.  This provides guarantees of what downstream consumers can expect from a topic; Consumers can be sure that every event in a particular topic will conform to a backwards compatible schema.
 
By default, topics in Kafka only have a single partition.  If a topic in the WMF production environment needs more partitions for throughput reasons, an Ops admin can increase the number of partitions.  Messages are produced to random partitions within a topic.
 
== Producers ==
 
=== MediaWiki ===
When equipped with the [[Mw:Extension:EventBus|EventBus extension]], MediaWiki can emit events for page edits (including creation), moves, deletes, undeletes, and changes to revision visibility, or anything else you might want to add.
 
=== ChangeProp ===
[[mw:Change_propagation|ChangeProp]] uses EventBus events to expire RESTBase cached articles, as well as various other updates.
 
=== Usage Example for MediaWiki Developers ===
 
For our example, we want to emit a new cool event from MediaWiki, perhaps a <code>mediawiki/widget/twiddle</code> event.  We'll assume that there is a [[mw:Manual:Hooks|MediaWiki hook]] called <code>WidgetTwiddle</code> that is fired when the widget is twiddled.
 
First, we'll design our new schema and add it to the [https://github.com/wikimedia/mediawiki-event-schemas mediawiki/event-schemas repository], and also configure a topic to accept events of that schema in the [https://github.com/wikimedia/mediawiki-event-schemas/blob/master/config/eventbus-topics.yaml eventbus-topics.yaml] config file.
 
=== Schema Design ===
Clone mediawiki/event-schemas from https://gerrit.wikimedia.org/r/#/admin/projects/mediawiki/event-schemas .  You'll be creating a new JSONSchema inside the <code>jsonschema/mediawiki</code> directory. Create a new file: <code>jsonschema/mediawiki/widget/twiddle/1.yaml</code>.
 
Edit this file to create your new schema.  Look around at the other schemas that already exist for good ideas on how to organize yours.  In order to encourage consistency, these event schemas are more tightly reviewed than the [https://meta.wikimedia.org/w/index.php?title=Special%3AAllPages&from=&to=&namespace=470 EventLogging ones] you might be used to.  At the very least, you'll need to add the <code>meta</code> schema object.  Our schema will look like this:
 
<syntaxhighlight lang="yaml">
title: mediawiki/widget/twiddle
description: >
  Represents a widget twiddle event.
type: object
properties:
  ### Meta data object.  All events schemas should have this.
  meta:
    type: object
    properties:
      topic:
        description: The queue topic name this message belongs to.
        type: string
      schema_uri:
        description: >
          The URI identifying the jsonschema for this event.  This may be just
          a short uri containing only the name and revision at the end of the
          URI path.  e.g. schema_name/12345 is acceptable.  This field
          is not required.
        type: string
      uri:
        description: The unique URI identifying the event.
        type: string
        format: uri
      request_id:
        description: The unique UUID v1 ID of the event derived from the X-Request-Id header.
        type: string
        pattern: '^[a-fA-F0-9]{8}(-[a-fA-F0-9]{4}){3}-[a-fA-F0-9]{12}$'
      id:
        description: The unique ID of this event; should match the dt field.
        type: string
        pattern: '^[a-fA-F0-9]{8}(-[a-fA-F0-9]{4}){3}-[a-fA-F0-9]{12}$'
      dt:
        description: The time stamp of the event, in ISO8601 format.
        type: string
        format: date-time
      domain:
        description: The domain the event pertains to.
        type: string
    required:
      - topic
      - uri
      - id
      - dt
      - domain
 
  ### Mediawiki entity fields.  All Mediawiki entity events should have these.
  database:
    description: The name of the wiki database this event belongs to.
    type: string
 
  performer:
    description: Represents the user that twiddled this widget.
    type: object
    properties:
      user_id:
        description: >
          The user id that twiddled the widget.  This is optional, and
          will not be present for anonymous users.
        type: integer
      user_text:
        description: The text representation of the user that twiddled the widget.
        type: string
      user_groups:
        description: A list of the groups this user belongs to.  E.g. bot, sysop etc.
        type: array
        items:
          type: string
      user_is_bot:
        description: >
          True if this user is considered to be a bot.  This is checked
          via the $user->isBot() method, which considers both user_groups
          and user permissions.
        type: boolean
    required:
      - user_text
      - user_groups
      - user_is_bot
 
  page_id:
    description: The page ID of the page this revision belongs to.
    type: integer
    minimum: 1
 
  page_title:
    description: The normalized title of the page this revision belongs to.
    type: string
 
  page_namespace:
    description: The namespace of the page this revision belongs to.
    type: integer
 
  widget_id:
    description: The ID of the widget being twiddled.
    type: integer
    minimum: 1
 
  widget_name:
    description: The name of the widget being twiddled.
    type: string
 
  widget_action:
    description: The twiddle action
    type: string
 
required:
  - meta
  - database
  - performer
  - widget_id
  - widget_name
  - widget_action
 
</syntaxhighlight>
 
<code>meta.topic</code> specifies what topic your event should go to.  This will probably be something close to the schema name.  Your schema reviewers will help you pick this.  We'll talk more about this in the next section.
 
<code>meta.schema_uri</code> is a relative URI naming your schema.  Ours will be mediawiki/widget/twiddle. 
 
<code>meta.uri</code>should be a unique URI representing the entity that the event is about.  This does not have to map to a reachable http URI.  In our case, this might be something like /widget/12345.
 
<code>meta.request_id</code> and <code>meta.id</code> will be generated for us by the EventBus MediaWiki extension.
 
<code>meta.dt</code> should represent the timestamp of the event in UTC [[wikipedia:ISO_8601|ISO-8601]] format, e.g 2017-06-07T08:43:15+00:00  If this is not provided, it will be generated for us.
 
<code>domain</code> is usually the wiki URI domain, e.g. en.wikipedia.org
 
The <code>meta</code> schema object is required for all schemas.  We try to keep common MediaWiki specific fields consistent across schemas, so take a look at other schemas to see if a MediaWiki data field you need for your schema has been named before.  Some conventions are documented at [https://github.com/wikimedia/mediawiki-event-schemas/tree/master/jsonschema/mediawiki MediaWiki Specific Event Schemas].
 
Fields are optional by default.  You should name as many required fields as you can, to help enforce consistency between different users of the schema.
 
=== Topic Config ===
If you are familiar with [[mw:Extension:EventLogging|EventLogging]], you know about schemas already.  In EventLogging, schemas do have a 'topic', but they are mapped directly to the schema name.  You don't control what the topic is.  This means that EventLogging schemas are not reusable.  Often, this is fine, but for some things, schemas should be reusable.  MediaWiki and other services in production use a <code>[https://github.com/wikimedia/mediawiki-event-schemas/blob/master/jsonschema/resource_change/1.yaml resource/change]</code> schema that is used to represent a change in any resource, not just MediaWiki based ones.  Topics allow us to send events of the same schema to different queues.  The word 'topic' comes from the [https://kafka.apache.org/ Apache Kafka] world, and is used by the EventBus service to know to which Kafka topic an event should be sent, but the concept applies generally.  You can think of a topic as a queue name.
 
The EventBus service controls which schemas are allowed to be produced to which topics via the [https://github.com/wikimedia/mediawiki-event-schemas/blob/master/config/eventbus-topics.yaml topics config file].  This mostly just maps topics to schema names.  We'll add a mapping for our new schema:<syntaxhighlight lang="yaml">
mediawiki.widget-twiddle:
    schema_name: mediawiki/widget/twiddle
 
</syntaxhighlight>Push your changes for review, and add someone in Services and/or Analytics (perhaps [[User:Mobrovac|Mobrovac]] and/or [[User:Ottomata|Ottomata]]) as reviewers.
 
=== Event Production from MediaWiki ===
Clone the [[mw:Extension:EventBus|EventBus extension]] from https://gerrit.wikimedia.org/r/#/admin/projects/mediawiki/extensions/EventBus.  We'll add a new handler function for the <code>WidgetTwiddle</code> hook.
 
Edit the <code>extension.json</code> config file and add an entry that maps your hook to the new function we are going to create:<syntaxhighlight lang="json">
"WidgetTwiddle": [
"EventBusHooks::onWidgetTwiddle"
]
</syntaxhighlight>Now edit the <code>Eventbus.hooks.php</code> file.  We'll add a new static function called <code>onWidgetTwiddle</code>.<syntaxhighlight lang="php">
/**
* Fires when a Widget is twiddled.
*
* @param Title $title title corresponding to the article the widget was on.
* @param Widget $widget object
* @param string $widgetAction
*/
public static function onWidgetTwiddle( Title $title, Widget $widget, $widgetAction ) {
global $wgDBname;
$events = [];
 
$performer = RequestContext::getMain()->getUser();
 
// Create a mediawiki page undelete event.
$attrs = [
// Common Mediawiki entity fields
'database'          => $wgDBname,
'performer'          => EventBus::createPerformerAttrs( $performer ),
 
// page entity fields
'page_id'            => $title->getArticleID(),
'page_title'        => $title->getPrefixedDBkey(),
'page_namespace'    => $title->getNamespace(),
'widget_id'          => $widget->getId(),
'widget_name'        => $widget->getName(),
'widget_action'      => $widgetAction
];
 
$events[] = EventBus::createEvent(
// This will end up being the meta.uri
$widget->getWidgetURI(),
// The topic this event should be produced to.
'mediawiki.widget-twiddle',
// This rest of the event attributes.
$attrs
);
 
DeferredUpdates::addCallableUpdate( function() use ( $events ) {
EventBus::getInstance()->send( $events );
} );
}
</syntaxhighlight>Push this change for review.  Once it and the changes to the mediawiki/event-schemas repository have been merged and deployed, your events will start flowing into Kafka.
 
Events are consumed into Hadoop, but querying them in Hive is not yet easy. ([[phab:T162610|T162610]] will help with this.) If you need these events in MySQL, like EventLogging, ask the Analytics team to help.
 
EventBus events can also be exposed publicly via [[EventStreams]].  If you think this would be useful, ask the Analytics team.
 
=== Direct Consumption from Kafka ===
You may consume your events using a Kafka client of your choice.  Shown here are examples using [https://github.com/edenhill/kafkacat kafkacat] and [https://github.com/Parsely/pykafka pykafka].
 
'''kafkacat'''
<syntaxhighlight lang="bash">
kafkacat -C -b deployment-kafka04.eqiad.wmflabs:9092 -t mediawiki.widget-twiddle
</syntaxhighlight>
 
'''pykafka'''
<syntaxhighlight lang="python">
from pykafka import KafkaClient
client = KafkaClient(hosts='kafka1001.eqiad.wmnet:9092,kafka1002.eqiad.wmnet:9092')
topic = client.topics['mediawiki.widget-twiddle']
consumer = topic.get_simple_consumer()
for page_edit_event in consumer:
print page_edit_event.value
</syntaxhighlight>
 
== MediaWiki Vagrant Development Environment ==
 
The EventBus service and MediaWIki extension can be installed by enabling the eventbus role in Vagrant:
  vagrant roles enable eventbus
  vagrant provision
 
The EventLogging server side codebase will be deployed to <code>/vagrant/srv/eventlogging</code>.  All EventBus events will be written out to <code>/vagrant/logs/eventbus.log</code>.  Try it out!  <code>tail -f /vagrant/logs/eventbus.log</code> and then edit a page.  You should see a <code>mediawiki/revision/create</code> event. 
 
=== Debugging and developing EventBus service in EventLogging codebase: ===
To activate eventlogging's virtualenv:
. /vagrant/srv/eventlogging/virtualenv/bin/activate
 
Command to start an extra eventbus process in vagrant for debugging the eventlogging codebase:
 
<source lang="bash">
. /vagrant/srv/eventlogging/virtualenv/bin/activate
eventlogging-service --port <PORT> --schemas-path /vagrant/srv/event-schemas/jsonschema --topic-config /vagrant/srv/event-schemas/config/eventbus-topics.yaml 'kafka:///localhost:9092?async=False&topic=datacenter1.{meta[topic]}' stdout://
</source>
 
 
To restart the main mediawiki vagrant eventbus service process:
  service eventlogging-service-eventbus restart
 
Process logs are in <code>/var/log/syslog</code>.  Events will be produced to Kafka and also output to <tt>/vagrant/logs/eventbus.log</tt>
 
 
Testing might involve deploying a different version of eventlogging
 
(virtualenv) >python setup.py build
(virtualenv) >python setup.py install
 
=== Posting events directly to EventBus service ===
 
  cat event.json
  {"comment": "Created page with \"woooo\"", "database": "wiki", "meta": {"domain": "dev.wiki.local.wmftest.net", "dt": "2017-06-07T19:53:08+00:00", "id": "ed98038c-4bba-11e7-b02e-0800278dc04d", "request_id": "0c4dc11e-0eed-4cba-94b6-071a48ae3189", "topic": "mediawiki.revision-create", "uri": "<nowiki>http://dev.wiki.local.wmftest.net:8080/wiki/Test1</nowiki>"}, "page_id": 3, "page_is_redirect": false, "page_namespace": 0, "page_title": "Test1", "performer": {"user_groups": ["*"], "user_is_bot": false, "user_text": "10.0.2.2"}, "rev_content_format": "wikitext", "rev_content_model": "wikitext", "rev_id": 4, "rev_len": 5, "rev_minor_edit": false, "rev_sha1": "oclx0411bem1x8t7qw1nk5rooeifcfp", "rev_timestamp": "2017-06-07T19:53:07Z"}
 
 
POST the event
      curl -H 'Content-Type: application/json' -X POST http://localhost:8085/v1/events -d @event.json
 
== References ==
<references/>
 
 
[[Category:Services]]

Latest revision as of 20:21, 20 September 2021

Redirect to:

Retrieved from "https://wikitech-static.wikimedia.org/w/index.php?title=Event_Platform/EventBus&oldid=583594"