Add an intracluster synchronization log for cluster repositories

Summary:
Depends on D19778. Ref T13216. See PHI943, PHI889, et al.

We currently have a push log and a pull log, but do not separately log intracluster synchronization events. We've encountered several specific cases where having this kind of log would be helpful:

  - In PHI943, an install was accidentally aborting locks early. Having timing information in the sync log would let us identify this more quickly.
  - In PHI889, an install hit an issue with `MaxStartups` configuration in `sshd`. A log would let us identify when this is an issue.
  - In PHI889, I floated a "push the linux kernel + fetch timeout" theory. A sync log would let us see sync/fetch timeouts to confirm this being a problem in practice.
  - A sync log will help us assess, develop, test, and monitor intracluster routing sync changes (likely those in T13211) in the future.

Some of these events are present in the pull log already, but only if they make it as far as running a `git upload-pack` subprocess (not the case with `MaxStartups` problems) -- and they can't record end-to-end timing.

No UI yet, I'll add that in a future change.

Test Plan:
  - Forced all operations to synchronize by adding `|| true` to the version check.
  - Pulled, got a sync log in the database.

Reviewers: amckinley

Reviewed By: amckinley

Subscribers: PHID-OPKG-gm6ozazyms6q6i22gyam

Maniphest Tasks: T13216

Differential Revision: https://secure.phabricator.com/D19779

resources/sql/autopatches/20181106.repo.01.sync.sql

@@ -0,0 +1,14 @@
+CREATE TABLE {$NAMESPACE}_repository.repository_syncevent (
+  id INT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
+  phid VARBINARY(64) NOT NULL,
+  repositoryPHID VARBINARY(64) NOT NULL,
+  epoch INT UNSIGNED NOT NULL,
+  devicePHID VARBINARY(64) NOT NULL,
+  fromDevicePHID VARBINARY(64) NOT NULL,
+  deviceVersion INT UNSIGNED,
+  fromDeviceVersion INT UNSIGNED,
+  resultType VARCHAR(32) NOT NULL COLLATE {$COLLATE_TEXT},
+  resultCode INT UNSIGNED NOT NULL,
+  syncWait BIGINT UNSIGNED NOT NULL,
+  properties LONGTEXT NOT NULL COLLATE {$COLLATE_TEXT}
+) ENGINE=InnoDB, COLLATE {$COLLATE_TEXT};

src/__phutil_library_map__.php

@@ -4163,6 +4163,9 @@ phutil_register_library_map(array(
     'PhabricatorRepositorySvnCommitChangeParserWorker' => 'applications/repository/worker/commitchangeparser/PhabricatorRepositorySvnCommitChangeParserWorker.php',
     'PhabricatorRepositorySvnCommitMessageParserWorker' => 'applications/repository/worker/commitmessageparser/PhabricatorRepositorySvnCommitMessageParserWorker.php',
     'PhabricatorRepositorySymbol' => 'applications/repository/storage/PhabricatorRepositorySymbol.php',
+    'PhabricatorRepositorySyncEvent' => 'applications/repository/storage/PhabricatorRepositorySyncEvent.php',
+    'PhabricatorRepositorySyncEventPHIDType' => 'applications/repository/phid/PhabricatorRepositorySyncEventPHIDType.php',
+    'PhabricatorRepositorySyncEventQuery' => 'applications/repository/query/PhabricatorRepositorySyncEventQuery.php',
     'PhabricatorRepositoryTestCase' => 'applications/repository/storage/__tests__/PhabricatorRepositoryTestCase.php',
     'PhabricatorRepositoryTransaction' => 'applications/repository/storage/PhabricatorRepositoryTransaction.php',
     'PhabricatorRepositoryTransactionQuery' => 'applications/repository/query/PhabricatorRepositoryTransactionQuery.php',
@@ -10111,6 +10114,12 @@ phutil_register_library_map(array(
     'PhabricatorRepositorySvnCommitChangeParserWorker' => 'PhabricatorRepositoryCommitChangeParserWorker',
     'PhabricatorRepositorySvnCommitMessageParserWorker' => 'PhabricatorRepositoryCommitMessageParserWorker',
     'PhabricatorRepositorySymbol' => 'PhabricatorRepositoryDAO',
+    'PhabricatorRepositorySyncEvent' => array(
+      'PhabricatorRepositoryDAO',
+      'PhabricatorPolicyInterface',
+    ),
+    'PhabricatorRepositorySyncEventPHIDType' => 'PhabricatorPHIDType',
+    'PhabricatorRepositorySyncEventQuery' => 'PhabricatorCursorPagedPolicyAwareQuery',
     'PhabricatorRepositoryTestCase' => 'PhabricatorTestCase',
     'PhabricatorRepositoryTransaction' => 'PhabricatorApplicationTransaction',
     'PhabricatorRepositoryTransactionQuery' => 'PhabricatorApplicationTransactionQuery',

src/applications/diffusion/protocol/DiffusionRepositoryClusterEngine.php

@@ -206,7 +206,10 @@ final class DiffusionRepositoryClusterEngine extends Phobject {
             }
           }
 
-          $this->synchronizeWorkingCopyFromDevices($fetchable);
+          $this->synchronizeWorkingCopyFromDevices(
+            $fetchable,
+            $this_version,
+            $max_version);
         } else {
           $this->synchronizeWorkingCopyFromRemote();
         }
@@ -653,7 +656,11 @@ final class DiffusionRepositoryClusterEngine extends Phobject {
   /**
    * @task internal
    */
-  private function synchronizeWorkingCopyFromDevices(array $device_phids) {
+  private function synchronizeWorkingCopyFromDevices(
+    array $device_phids,
+    $local_version,
+    $remote_version) {
+
     $repository = $this->getRepository();
 
     $service = $repository->loadAlmanacService();
@@ -694,7 +701,10 @@ final class DiffusionRepositoryClusterEngine extends Phobject {
     $caught = null;
     foreach ($fetchable as $binding) {
       try {
-        $this->synchronizeWorkingCopyFromBinding($binding);
+        $this->synchronizeWorkingCopyFromBinding(
+          $binding,
+          $local_version,
+          $remote_version);
         $caught = null;
         break;
       } catch (Exception $ex) {
@@ -711,14 +721,17 @@ final class DiffusionRepositoryClusterEngine extends Phobject {
   /**
    * @task internal
    */
-  private function synchronizeWorkingCopyFromBinding($binding) {
+  private function synchronizeWorkingCopyFromBinding(
+    AlmanacBinding $binding,
+    $local_version,
+    $remote_version) {
+
     $repository = $this->getRepository();
     $device = AlmanacKeys::getLiveDevice();
 
     $this->logLine(
       pht(
-        'Synchronizing this device ("%s") from cluster leader ("%s") before '.
-        'read.',
+        'Synchronizing this device ("%s") from cluster leader ("%s").',
         $device->getName(),
         $binding->getDevice()->getName()));
 
@@ -746,17 +759,60 @@ final class DiffusionRepositoryClusterEngine extends Phobject {
 
     $future->setCWD($local_path);
 
+    $log = PhabricatorRepositorySyncEvent::initializeNewEvent()
+      ->setRepositoryPHID($repository->getPHID())
+      ->setEpoch(PhabricatorTime::getNow())
+      ->setDevicePHID($device->getPHID())
+      ->setFromDevicePHID($binding->getDevice()->getPHID())
+      ->setDeviceVersion($local_version)
+      ->setFromDeviceVersion($remote_version);
+
+    $sync_start = microtime(true);
+
     try {
       $future->resolvex();
     } catch (Exception $ex) {
+      $sync_end = microtime(true);
+      $log->setSyncWait((int)(1000000 * ($sync_end - $sync_start)));
+
+      if ($ex instanceof CommandException) {
+        if ($future->getWasKilledByTimeout()) {
+          $result_type = PhabricatorRepositorySyncEvent::RESULT_TIMEOUT;
+        } else {
+          $result_type = PhabricatorRepositorySyncEvent::RESULT_ERROR;
+        }
+
+       $log
+         ->setResultCode($ex->getError())
+         ->setResultType($result_type)
+         ->setProperty('stdout', $ex->getStdout())
+         ->setProperty('stderr', $ex->getStderr());
+      } else {
+        $log
+          ->setResultCode(1)
+          ->setResultType(PhabricatorRepositorySyncEvent::RESULT_EXCEPTION)
+          ->setProperty('message', $ex->getMessage());
+      }
+
+      $log->save();
+
       $this->logLine(
         pht(
           'Synchronization of "%s" from leader "%s" failed: %s',
           $device->getName(),
           $binding->getDevice()->getName(),
           $ex->getMessage()));
+
       throw $ex;
     }
+
+    $sync_end = microtime(true);
+
+    $log
+      ->setSyncWait((int)(1000000 * ($sync_end - $sync_start)))
+      ->setResultCode(0)
+      ->setResultType(PhabricatorRepositorySyncEvent::RESULT_SYNC)
+      ->save();
   }
 
 

src/applications/repository/phid/PhabricatorRepositorySyncEventPHIDType.php

@@ -0,0 +1,39 @@
+<?php
+
+final class PhabricatorRepositorySyncEventPHIDType extends PhabricatorPHIDType {
+
+  const TYPECONST = 'SYNE';
+
+  public function getTypeName() {
+    return pht('Sync Event');
+  }
+
+  public function newObject() {
+    return new PhabricatorRepositorySyncEvent();
+  }
+
+  public function getPHIDTypeApplicationClass() {
+    return 'PhabricatorDiffusionApplication';
+  }
+
+  protected function buildQueryForObjects(
+    PhabricatorObjectQuery $query,
+    array $phids) {
+
+    return id(new PhabricatorRepositorySyncEventQuery())
+      ->withPHIDs($phids);
+  }
+
+  public function loadHandles(
+    PhabricatorHandleQuery $query,
+    array $handles,
+    array $objects) {
+
+    foreach ($handles as $phid => $handle) {
+      $event = $objects[$phid];
+
+      $handle->setName(pht('Sync Event %d', $event->getID()));
+    }
+  }
+
+}

src/applications/repository/query/PhabricatorRepositorySyncEventQuery.php

@@ -0,0 +1,115 @@
+<?php
+
+final class PhabricatorRepositorySyncEventQuery
+  extends PhabricatorCursorPagedPolicyAwareQuery {
+
+  private $ids;
+  private $phids;
+  private $repositoryPHIDs;
+  private $epochMin;
+  private $epochMax;
+
+  public function withIDs(array $ids) {
+    $this->ids = $ids;
+    return $this;
+  }
+
+  public function withPHIDs(array $phids) {
+    $this->phids = $phids;
+    return $this;
+  }
+
+  public function withRepositoryPHIDs(array $repository_phids) {
+    $this->repositoryPHIDs = $repository_phids;
+    return $this;
+  }
+
+  public function withEpochBetween($min, $max) {
+    $this->epochMin = $min;
+    $this->epochMax = $max;
+    return $this;
+  }
+
+  public function newResultObject() {
+    return new PhabricatorRepositoryPullEvent();
+  }
+
+  protected function loadPage() {
+    return $this->loadStandardPage($this->newResultObject());
+  }
+
+  protected function willFilterPage(array $events) {
+    $repository_phids = mpull($events, 'getRepositoryPHID');
+    $repository_phids = array_filter($repository_phids);
+
+    if ($repository_phids) {
+      $repositories = id(new PhabricatorRepositoryQuery())
+        ->setViewer($this->getViewer())
+        ->withPHIDs($repository_phids)
+        ->execute();
+      $repositories = mpull($repositories, null, 'getPHID');
+    } else {
+      $repositories = array();
+    }
+
+    foreach ($events as $key => $event) {
+      $phid = $event->getRepositoryPHID();
+
+      if (empty($repositories[$phid])) {
+        unset($events[$key]);
+        $this->didRejectResult($event);
+        continue;
+      }
+
+      $event->attachRepository($repositories[$phid]);
+    }
+
+    return $events;
+  }
+
+  protected function buildWhereClauseParts(AphrontDatabaseConnection $conn) {
+    $where = parent::buildWhereClauseParts($conn);
+
+    if ($this->ids !== null) {
+      $where[] = qsprintf(
+        $conn,
+        'id IN (%Ld)',
+        $this->ids);
+    }
+
+    if ($this->phids !== null) {
+      $where[] = qsprintf(
+        $conn,
+        'phid IN (%Ls)',
+        $this->phids);
+    }
+
+    if ($this->repositoryPHIDs !== null) {
+      $where[] = qsprintf(
+        $conn,
+        'repositoryPHID IN (%Ls)',
+        $this->repositoryPHIDs);
+    }
+
+    if ($this->epochMin !== null) {
+      $where[] = qsprintf(
+        $conn,
+        'epoch >= %d',
+        $this->epochMin);
+    }
+
+    if ($this->epochMax !== null) {
+      $where[] = qsprintf(
+        $conn,
+        'epoch <= %d',
+        $this->epochMax);
+    }
+
+    return $where;
+  }
+
+  public function getQueryApplicationClass() {
+    return 'PhabricatorDiffusionApplication';
+  }
+
+}

src/applications/repository/storage/PhabricatorRepositorySyncEvent.php

@@ -0,0 +1,99 @@
+<?php
+
+final class PhabricatorRepositorySyncEvent
+  extends PhabricatorRepositoryDAO
+  implements PhabricatorPolicyInterface {
+
+  protected $repositoryPHID;
+  protected $epoch;
+  protected $devicePHID;
+  protected $fromDevicePHID;
+  protected $deviceVersion;
+  protected $fromDeviceVersion;
+  protected $resultType;
+  protected $resultCode;
+  protected $syncWait;
+  protected $properties = array();
+
+  private $repository = self::ATTACHABLE;
+
+  const RESULT_SYNC = 'sync';
+  const RESULT_ERROR = 'error';
+  const RESULT_TIMEOUT = 'timeout';
+  const RESULT_EXCEPTION = 'exception';
+
+  public static function initializeNewEvent() {
+    return new self();
+  }
+
+  protected function getConfiguration() {
+    return array(
+      self::CONFIG_AUX_PHID => true,
+      self::CONFIG_TIMESTAMPS => false,
+      self::CONFIG_SERIALIZATION => array(
+        'properties' => self::SERIALIZATION_JSON,
+      ),
+      self::CONFIG_COLUMN_SCHEMA => array(
+        'deviceVersion' => 'uint32?',
+        'fromDeviceVersion' => 'uint32?',
+        'resultType' => 'text32',
+        'resultCode' => 'uint32',
+        'syncWait' => 'uint64',
+      ),
+      self::CONFIG_KEY_SCHEMA => array(
+        'key_repository' => array(
+          'columns' => array('repositoryPHID'),
+        ),
+        'key_epoch' => array(
+          'columns' => array('epoch'),
+        ),
+      ),
+    ) + parent::getConfiguration();
+  }
+
+  public function getPHIDType() {
+    return PhabricatorRepositorySyncEventPHIDType::TYPECONST;
+  }
+
+  public function attachRepository(PhabricatorRepository $repository) {
+    $this->repository = $repository;
+    return $this;
+  }
+
+  public function getRepository() {
+    return $this->assertAttached($this->repository);
+  }
+
+  public function setProperty($key, $value) {
+    $this->properites[$key] = $value;
+    return $this;
+  }
+
+  public function getProperty($key, $default = null) {
+    return idx($this->properties, $key, $default);
+  }
+
+/* -(  PhabricatorPolicyInterface  )----------------------------------------- */
+
+
+  public function getCapabilities() {
+    return array(
+      PhabricatorPolicyCapability::CAN_VIEW,
+    );
+  }
+
+  public function getPolicy($capability) {
+    return $this->getRepository()->getPolicy($capability);
+  }
+
+  public function hasAutomaticCapability($capability, PhabricatorUser $viewer) {
+    return $this->getRepository()->hasAutomaticCapability($capability, $viewer);
+  }
+
+  public function describeAutomaticCapability($capability) {
+    return pht(
+      "A repository's sync events are visible to users who can see the ".
+      "repository.");
+  }
+
+}