From 621487f6039b659927d1f73ed3a99c588a4a173d Mon Sep 17 00:00:00 2001
From: Aaron Schulz <aschulz@wikimedia.org>
Date: Tue, 29 Dec 2015 11:45:23 -0800
Subject: [PATCH] Make CDN purges send EventRelayer events

* Also create EventRelayerGroup class and config

Bug: T97562
Change-Id: Id7c9b0deb0ffd21277572a8e6407f87be6931a4d
---
 RELEASE-NOTES-1.27                   |  2 +
 autoload.php                         |  1 +
 includes/DefaultSettings.php         | 19 +++++++++
 includes/EventRelayerGroup.php       | 58 ++++++++++++++++++++++++++++
 includes/deferred/CdnCacheUpdate.php | 12 ++++++
 5 files changed, 92 insertions(+)
 create mode 100644 includes/EventRelayerGroup.php

diff --git a/RELEASE-NOTES-1.27 b/RELEASE-NOTES-1.27
index 04bab03145..9ff9ab0fbf 100644
--- a/RELEASE-NOTES-1.27
+++ b/RELEASE-NOTES-1.27
@@ -163,6 +163,8 @@ production.
 * Added $wgAuthenticationTokenVersion, which if non-null prevents the
   user_token database field from being exposed in cookies. Setting this would
   be a good idea, but will log out all current sessions.
+* $wgEventRelayerConfig was added, for managing PubSub event relay configuration,
+  specifically for reliable CDN url purges.
 
 === External library changes in 1.27 ===
 
diff --git a/autoload.php b/autoload.php
index 937b8c9d34..0a9d80c6f1 100644
--- a/autoload.php
+++ b/autoload.php
@@ -395,6 +395,7 @@ $wgAutoloadLocalClasses = array(
 	'EraseArchivedFile' => __DIR__ . '/maintenance/eraseArchivedFile.php',
 	'ErrorPageError' => __DIR__ . '/includes/exception/ErrorPageError.php',
 	'EventRelayer' => __DIR__ . '/includes/libs/eventrelayer/EventRelayer.php',
+	'EventRelayerGroup' => __DIR__ . '/includes/EventRelayerGroup.php',
 	'EventRelayerMCRD' => __DIR__ . '/includes/libs/eventrelayer/EventRelayerMCRD.php',
 	'EventRelayerNull' => __DIR__ . '/includes/libs/eventrelayer/EventRelayer.php',
 	'Exif' => __DIR__ . '/includes/media/Exif.php',
diff --git a/includes/DefaultSettings.php b/includes/DefaultSettings.php
index b9e5d2f2c1..a6a0c75b54 100644
--- a/includes/DefaultSettings.php
+++ b/includes/DefaultSettings.php
@@ -7965,6 +7965,25 @@ $wgPopularPasswordFile = __DIR__ . '/../serialized/commonpasswords.cdb';
  */
 $wgMaxUserDBWriteDuration = false;
 
+/**
+ * Mapping of event channels to EventRelayer configuration.
+ *
+ * By setting up a PubSub system (like Kafka) and enabling a corresponding EventRelayer class
+ * that uses it, MediaWiki can broadcast events to all subscribers. Certain features like WAN
+ * cache purging and CDN cache purging will emit events to this system. Appropriate listers can
+ * subscribe to the channel and take actions based on the events. For example, a local daemon
+ * can run on each CDN cache node and perfom local purges based on the URL purge channel events.
+ *
+ * The 'default' channel is for all channels without an explicit entry here.
+ *
+ * @since 1.27
+ */
+$wgEventRelayerConfig = array(
+	'default' => array(
+		'class' => 'EventRelayerNull',
+	)
+);
+
 /**
  * For really cool vim folding this needs to be at the end:
  * vim: foldmarker=@{,@} foldmethod=marker
diff --git a/includes/EventRelayerGroup.php b/includes/EventRelayerGroup.php
new file mode 100644
index 0000000000..45e9be91e9
--- /dev/null
+++ b/includes/EventRelayerGroup.php
@@ -0,0 +1,58 @@
+<?php
+/**
+ * Factory class for spawning EventRelayer objects using configuration
+ *
+ * @author Aaron Schulz
+ * @since 1.27
+ */
+class EventRelayerGroup {
+	/** @var array[] */
+	protected $configByChannel = array();
+
+	/** @var EventRelayer[] */
+	protected $relayers = array();
+
+	/** @var EventRelayerGroup */
+	protected static $instance = null;
+
+	/**
+	 * @param Config $config
+	 */
+	protected function __constuct( Config $config ) {
+		$this->configByChannel = $config->get( 'EventRelayerConfig' );
+	}
+
+	/**
+	 * @return EventRelayerGroup
+	 */
+	public static function singleton() {
+		if ( !self::$instance ) {
+			self::$instance = new self( RequestContext::getMain()->getConfig() );
+		}
+
+		return self::$instance;
+	}
+
+	/**
+	 * @param string $channel
+	 * @return EventRelayer Relayer instance that handles the given channel
+	 */
+	public function getRelayer( $channel ) {
+		$channelKey = isset( $this->configByChannel[$channel] )
+			? $channel
+			: 'default';
+
+		if ( !isset( $this->relayers[$channelKey] ) ) {
+			if ( !isset( $this->configByChannel[$channelKey] ) ) {
+				throw new UnexpectedValueException( "No config for '$channelKey'" );
+			}
+
+			$config = $this->configByChannel[$channelKey];
+			$class = $config['class'];
+
+			$this->relayers[$channelKey] = new $class( $config );
+		}
+
+		return $this->relayers[$channelKey];
+	}
+}
diff --git a/includes/deferred/CdnCacheUpdate.php b/includes/deferred/CdnCacheUpdate.php
index 9f7d8caec7..b5e323bbc5 100644
--- a/includes/deferred/CdnCacheUpdate.php
+++ b/includes/deferred/CdnCacheUpdate.php
@@ -108,10 +108,22 @@ class CdnCacheUpdate implements DeferrableUpdate, MergeableUpdate {
 
 		wfDebugLog( 'squid', __METHOD__ . ': ' . implode( ' ', $urlArr ) );
 
+		// Reliably broadcast the purge to all edge nodes
+		$relayer = EventRelayerGroup::singleton()->getRelayer( 'cdn-url-purges' );
+		$relayer->notify(
+			'cdn-url-purges',
+			array(
+				'urls' => array_values( $urlArr ), // JSON array
+				'timestamp' => microtime( true )
+			)
+		);
+
+		// Send lossy UDP broadcasting if enabled
 		if ( $wgHTCPRouting ) {
 			self::HTCPPurge( $urlArr );
 		}
 
+		// Do direct server purges if enabled (this does not scale very well)
 		if ( $wgSquidServers ) {
 			// Maximum number of parallel connections per squid
 			$maxSocketsPerSquid = 8;
-- 
2.20.1