ArchivingStatus.php
2 years ago
Loader.php
8 months ago
LoaderLock.php
1 year ago
Parameters.php
8 months ago
PluginsArchiver.php
1 year ago
PluginsArchiverException.php
2 years ago
Record.php
1 year ago
RecordBuilder.php
8 months ago
Rules.php
1 year ago
RecordBuilder.php
274 lines
| 1 | <?php |
| 2 | |
| 3 | /** |
| 4 | * Matomo - free/libre analytics platform |
| 5 | * |
| 6 | * @link https://matomo.org |
| 7 | * @license https://www.gnu.org/licenses/gpl-3.0.html GPL v3 or later |
| 8 | */ |
| 9 | namespace Piwik\ArchiveProcessor; |
| 10 | |
| 11 | use Piwik\ArchiveProcessor; |
| 12 | use Piwik\Common; |
| 13 | use Piwik\DataTable; |
| 14 | use Piwik\Piwik; |
| 15 | /** |
| 16 | * Inherit from this class to define archiving logic for one or more records. |
| 17 | */ |
| 18 | abstract class RecordBuilder |
| 19 | { |
| 20 | /** |
| 21 | * @var int |
| 22 | */ |
| 23 | protected $maxRowsInTable; |
| 24 | /** |
| 25 | * @var int |
| 26 | */ |
| 27 | protected $maxRowsInSubtable; |
| 28 | /** |
| 29 | * @var string|int|null |
| 30 | */ |
| 31 | protected $columnToSortByBeforeTruncation; |
| 32 | /** |
| 33 | * @var array|null |
| 34 | */ |
| 35 | protected $columnAggregationOps; |
| 36 | /** |
| 37 | * @var array|null |
| 38 | */ |
| 39 | protected $columnToRenameAfterAggregation; |
| 40 | /** |
| 41 | * @param int|null $maxRowsInTable |
| 42 | * @param int|null $maxRowsInSubtable |
| 43 | * @param string|null $columnToSortByBeforeTruncation |
| 44 | * @param array|null $columnAggregationOps |
| 45 | */ |
| 46 | public function __construct(?int $maxRowsInTable = null, ?int $maxRowsInSubtable = null, ?string $columnToSortByBeforeTruncation = null, ?array $columnAggregationOps = null, ?array $columnToRenameAfterAggregation = null) |
| 47 | { |
| 48 | $this->maxRowsInTable = $maxRowsInTable; |
| 49 | $this->maxRowsInSubtable = $maxRowsInSubtable; |
| 50 | $this->columnToSortByBeforeTruncation = $columnToSortByBeforeTruncation; |
| 51 | $this->columnAggregationOps = $columnAggregationOps; |
| 52 | $this->columnToRenameAfterAggregation = $columnToRenameAfterAggregation; |
| 53 | } |
| 54 | public function isEnabled(ArchiveProcessor $archiveProcessor) : bool |
| 55 | { |
| 56 | return \true; |
| 57 | } |
| 58 | /** |
| 59 | * Uses the protected `aggregate()` function to build records by aggregating log table data directly, then |
| 60 | * inserts them as archive data. |
| 61 | * |
| 62 | * @param ArchiveProcessor $archiveProcessor |
| 63 | * @return void |
| 64 | */ |
| 65 | public function buildFromLogs(ArchiveProcessor $archiveProcessor) : void |
| 66 | { |
| 67 | if (!$this->isEnabled($archiveProcessor)) { |
| 68 | return; |
| 69 | } |
| 70 | $recordsBuilt = $this->getRecordMetadata($archiveProcessor); |
| 71 | $recordMetadataByName = []; |
| 72 | foreach ($recordsBuilt as $recordMetadata) { |
| 73 | if (!$recordMetadata instanceof \Piwik\ArchiveProcessor\Record) { |
| 74 | continue; |
| 75 | } |
| 76 | $recordMetadataByName[$recordMetadata->getName()] = $recordMetadata; |
| 77 | } |
| 78 | $numericRecords = []; |
| 79 | $records = $this->aggregate($archiveProcessor); |
| 80 | foreach ($records as $recordName => $recordValue) { |
| 81 | if (empty($recordMetadataByName[$recordName])) { |
| 82 | if ($recordValue instanceof DataTable) { |
| 83 | Common::destroy($recordValue); |
| 84 | } |
| 85 | continue; |
| 86 | } |
| 87 | if ($recordValue instanceof DataTable) { |
| 88 | $record = $recordMetadataByName[$recordName]; |
| 89 | $maxRowsInTable = $record->getMaxRowsInTable() ?? $this->maxRowsInTable; |
| 90 | $maxRowsInSubtable = $record->getMaxRowsInSubtable() ?? $this->maxRowsInSubtable; |
| 91 | $columnToSortByBeforeTruncation = $record->getColumnToSortByBeforeTruncation() ?? $this->columnToSortByBeforeTruncation; |
| 92 | $this->insertBlobRecord($archiveProcessor, $recordName, $recordValue, $maxRowsInTable, $maxRowsInSubtable, $columnToSortByBeforeTruncation); |
| 93 | Common::destroy($recordValue); |
| 94 | } else { |
| 95 | // collect numeric records so we can insert them all at once |
| 96 | $numericRecords[$recordName] = $recordValue; |
| 97 | } |
| 98 | } |
| 99 | unset($records); |
| 100 | if (!empty($numericRecords)) { |
| 101 | $archiveProcessor->insertNumericRecords($numericRecords); |
| 102 | } |
| 103 | } |
| 104 | /** |
| 105 | * Builds records for non-day periods by aggregating day records together, then inserts |
| 106 | * them as archive data. |
| 107 | * |
| 108 | * @param ArchiveProcessor $archiveProcessor |
| 109 | * @return void |
| 110 | */ |
| 111 | public function buildForNonDayPeriod(ArchiveProcessor $archiveProcessor) : void |
| 112 | { |
| 113 | if (!$this->isEnabled($archiveProcessor)) { |
| 114 | return; |
| 115 | } |
| 116 | $requestedReports = $archiveProcessor->getParams()->getArchiveOnlyReportAsArray(); |
| 117 | $foundRequestedReports = $archiveProcessor->getParams()->getFoundRequestedReports(); |
| 118 | $recordsBuilt = $this->getRecordMetadata($archiveProcessor); |
| 119 | $numericRecords = array_filter($recordsBuilt, function (\Piwik\ArchiveProcessor\Record $r) { |
| 120 | return $r->getType() == \Piwik\ArchiveProcessor\Record::TYPE_NUMERIC; |
| 121 | }); |
| 122 | $blobRecords = array_filter($recordsBuilt, function (\Piwik\ArchiveProcessor\Record $r) { |
| 123 | return $r->getType() == \Piwik\ArchiveProcessor\Record::TYPE_BLOB; |
| 124 | }); |
| 125 | $aggregatedCounts = []; |
| 126 | // make sure if there are requested numeric records that depend on blob records, that the blob records will be archived first |
| 127 | foreach ($numericRecords as $record) { |
| 128 | if (empty($record->getCountOfRecordName()) || !in_array($record->getName(), $requestedReports)) { |
| 129 | continue; |
| 130 | } |
| 131 | $dependentRecordName = $record->getCountOfRecordName(); |
| 132 | if (!in_array($dependentRecordName, $requestedReports)) { |
| 133 | $requestedReports[] = $dependentRecordName; |
| 134 | } |
| 135 | // we need to aggregate the blob record to get the count, so even if it's found, we must re-aggregate it |
| 136 | // TODO: this could potentially be optimized away, but it would be non-trivial given the current ArchiveProcessor API |
| 137 | $indexInFoundRecords = array_search($dependentRecordName, $foundRequestedReports); |
| 138 | if ($indexInFoundRecords !== \false) { |
| 139 | unset($foundRequestedReports[$indexInFoundRecords]); |
| 140 | } |
| 141 | } |
| 142 | foreach ($blobRecords as $record) { |
| 143 | if (!empty($requestedReports) && (!in_array($record->getName(), $requestedReports) || in_array($record->getName(), $foundRequestedReports))) { |
| 144 | continue; |
| 145 | } |
| 146 | $maxRowsInTable = $record->getMaxRowsInTable() ?? $this->maxRowsInTable; |
| 147 | $maxRowsInSubtable = $record->getMaxRowsInSubtable() ?? $this->maxRowsInSubtable; |
| 148 | $columnToSortByBeforeTruncation = $record->getColumnToSortByBeforeTruncation() ?? $this->columnToSortByBeforeTruncation; |
| 149 | $columnToRenameAfterAggregation = $record->getColumnToRenameAfterAggregation() ?? $this->columnToRenameAfterAggregation; |
| 150 | $columnAggregationOps = $record->getBlobColumnAggregationOps() ?? $this->columnAggregationOps; |
| 151 | // only do recursive row count if there is a numeric record that depends on it |
| 152 | $countRecursiveRows = \false; |
| 153 | foreach ($numericRecords as $numeric) { |
| 154 | if ($numeric->getCountOfRecordName() == $record->getName() && $numeric->getCountOfRecordNameIsRecursive()) { |
| 155 | $countRecursiveRows = \true; |
| 156 | break; |
| 157 | } |
| 158 | } |
| 159 | $counts = $archiveProcessor->aggregateDataTableRecords($record->getName(), $maxRowsInTable, $maxRowsInSubtable, $columnToSortByBeforeTruncation, $columnAggregationOps, $columnToRenameAfterAggregation, $countRecursiveRows); |
| 160 | $aggregatedCounts = array_merge($aggregatedCounts, $counts); |
| 161 | } |
| 162 | if (!empty($numericRecords)) { |
| 163 | // handle metrics that are aggregated using metric values from child periods |
| 164 | $autoAggregateMetrics = array_filter($numericRecords, function (\Piwik\ArchiveProcessor\Record $r) { |
| 165 | return empty($r->getCountOfRecordName()); |
| 166 | }); |
| 167 | $autoAggregateMetrics = array_map(function (\Piwik\ArchiveProcessor\Record $r) { |
| 168 | return $r->getName(); |
| 169 | }, $autoAggregateMetrics); |
| 170 | if (!empty($requestedReports)) { |
| 171 | $autoAggregateMetrics = array_filter($autoAggregateMetrics, function ($name) use($requestedReports, $foundRequestedReports) { |
| 172 | return in_array($name, $requestedReports) && !in_array($name, $foundRequestedReports); |
| 173 | }); |
| 174 | } |
| 175 | $autoAggregateMetrics = array_values($autoAggregateMetrics); |
| 176 | if (!empty($autoAggregateMetrics)) { |
| 177 | $archiveProcessor->aggregateNumericMetrics($autoAggregateMetrics, $this->columnAggregationOps); |
| 178 | } |
| 179 | // handle metrics that are set to counts of blob records |
| 180 | $recordCountMetricValues = []; |
| 181 | $recordCountMetrics = array_filter($numericRecords, function (\Piwik\ArchiveProcessor\Record $r) { |
| 182 | return !empty($r->getCountOfRecordName()); |
| 183 | }); |
| 184 | foreach ($recordCountMetrics as $record) { |
| 185 | $dependentRecordName = $record->getCountOfRecordName(); |
| 186 | if (empty($aggregatedCounts[$dependentRecordName])) { |
| 187 | continue; |
| 188 | // dependent record not archived, so skip this metric |
| 189 | } |
| 190 | $count = $aggregatedCounts[$dependentRecordName]; |
| 191 | if ($record->getCountOfRecordNameIsRecursive()) { |
| 192 | $recordCountMetricValues[$record->getName()] = $count['recursive']; |
| 193 | } else { |
| 194 | $recordCountMetricValues[$record->getName()] = $count['level0']; |
| 195 | } |
| 196 | $transform = $record->getMultiplePeriodTransform(); |
| 197 | if (!empty($transform)) { |
| 198 | $recordCountMetricValues[$record->getName()] = $transform($recordCountMetricValues[$record->getName()], $count); |
| 199 | } |
| 200 | } |
| 201 | if (!empty($recordCountMetricValues)) { |
| 202 | $archiveProcessor->insertNumericRecords($recordCountMetricValues); |
| 203 | } |
| 204 | } |
| 205 | } |
| 206 | /** |
| 207 | * Returns metadata for records primarily used when aggregating over non-day periods. Every numeric/blob |
| 208 | * record your RecordBuilder creates should have an associated piece of record metadata. |
| 209 | * |
| 210 | * @return Record[] |
| 211 | */ |
| 212 | public abstract function getRecordMetadata(ArchiveProcessor $archiveProcessor) : array; |
| 213 | /** |
| 214 | * Derived classes should define this method to aggregate log data for a single day and return the records |
| 215 | * to store indexed by record names. |
| 216 | * |
| 217 | * @return array<string, DataTable|int|float|string> Record values indexed by their record name, eg, `['MyPlugin_MyRecord' => new DataTable()]` |
| 218 | */ |
| 219 | protected abstract function aggregate(ArchiveProcessor $archiveProcessor) : array; |
| 220 | protected function insertBlobRecord(ArchiveProcessor $archiveProcessor, string $recordName, DataTable $record, ?int $maxRowsInTable, ?int $maxRowsInSubtable, ?string $columnToSortByBeforeTruncation) : void |
| 221 | { |
| 222 | $serialized = $record->getSerialized($maxRowsInTable ?: $this->maxRowsInTable, $maxRowsInSubtable ?: $this->maxRowsInSubtable, $columnToSortByBeforeTruncation ?: $this->columnToSortByBeforeTruncation); |
| 223 | $archiveProcessor->insertBlobRecord($recordName, $serialized); |
| 224 | unset($serialized); |
| 225 | } |
| 226 | public function getMaxRowsInTable() : ?int |
| 227 | { |
| 228 | return $this->maxRowsInTable; |
| 229 | } |
| 230 | public function getMaxRowsInSubtable() : ?int |
| 231 | { |
| 232 | return $this->maxRowsInSubtable; |
| 233 | } |
| 234 | public function getColumnToSortByBeforeTruncation() : ?string |
| 235 | { |
| 236 | return $this->columnToSortByBeforeTruncation; |
| 237 | } |
| 238 | public function getPluginName() : string |
| 239 | { |
| 240 | return Piwik::getPluginNameOfMatomoClass(get_class($this)); |
| 241 | } |
| 242 | /** |
| 243 | * Returns an extra hint for LogAggregator to add to log aggregation SQL. Can be overridden if you'd |
| 244 | * like the origin hint to have more information. |
| 245 | * |
| 246 | * @return string |
| 247 | */ |
| 248 | public function getQueryOriginHint() : string |
| 249 | { |
| 250 | $recordBuilderName = get_class($this); |
| 251 | $recordBuilderName = explode('\\', $recordBuilderName); |
| 252 | return end($recordBuilderName); |
| 253 | } |
| 254 | /** |
| 255 | * Returns true if at least one of the given reports is handled by this RecordBuilder instance |
| 256 | * when invoked with the given ArchiveProcessor. |
| 257 | * |
| 258 | * @param ArchiveProcessor $archiveProcessor Archiving parameters, like idSite, can influence the list of |
| 259 | * all records a RecordBuilder produces, so it is required here. |
| 260 | * @param string[] $requestedReports The list of requested reports to check for. |
| 261 | * @return bool |
| 262 | */ |
| 263 | public function isBuilderForAtLeastOneOf(ArchiveProcessor $archiveProcessor, array $requestedReports) : bool |
| 264 | { |
| 265 | $recordMetadata = $this->getRecordMetadata($archiveProcessor); |
| 266 | foreach ($recordMetadata as $record) { |
| 267 | if (in_array($record->getName(), $requestedReports)) { |
| 268 | return \true; |
| 269 | } |
| 270 | } |
| 271 | return \false; |
| 272 | } |
| 273 | } |
| 274 |