Skip to content

Commit

Permalink
wip
Browse files Browse the repository at this point in the history
  • Loading branch information
@jessarcher
jessarcher committed Dec 5, 2023
1 parent 44001c6 commit 945991e
Showing 1 changed file with 108 additions and 55 deletions.
163 changes: 108 additions & 55 deletions src/Storage/DatabaseStorage.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,17 +65,17 @@ public function store(Collection $items): void
];

$this
->aggregateAttributes($entries->filter->isCount(), $periods, 'count')
->aggregateCounts($entries->filter->isCount(), $periods)
->chunk($this->config->get('pulse.storage.database.chunk'))
->each(fn ($chunk) => $this->upsertCount($chunk->all()));

$this
->aggregateAttributes($entries->filter->isMax(), $periods, 'max')
->aggregateMaximums($entries->filter->isMax(), $periods)
->chunk($this->config->get('pulse.storage.database.chunk'))
->each(fn ($chunk) => $this->upsertMax($chunk->all()));

$this

Check failure on line 77 in src/Storage/DatabaseStorage.php

View workflow job for this annotation

GitHub Actions / Static Analysis 

Method Laravel\Pulse\Storage\DatabaseStorage::aggregateAverages() invoked with 2 parameters, 1 required.
->aggregateAttributes($entries->filter->isAvg(), $periods, 'avg')
->aggregateAverages($entries->filter->isAvg(), $periods)
->chunk($this->config->get('pulse.storage.database.chunk'))
->each(fn ($chunk) => $this->upsertAvg($chunk->all()));

Expand Down Expand Up @@ -213,80 +213,133 @@ protected function upsert(array $values, string $onDuplicateKeyClause): bool
}

/**
* Get the aggregate attributes for the collection.
* Get the count aggregates
*
* @param \Illuminate\Support\Collection<int, \Laravel\Pulse\Entry> $entries
* @param list<int> $periods
* @return \Illuminate\Support\LazyCollection<int, AggregateRow>
* @return \Illuminate\Support\Collection<int, AggregateRow>
*/
protected function aggregateAttributes(Collection $entries, array $periods, string $aggregateSuffix): Collection
protected function aggregateCounts(Collection $entries, array $periods): Collection
{
$aggregates = LazyCollection::make(function () use ($entries, $periods, $aggregateSuffix) {
foreach ($entries as $entry) {
foreach ($periods as $period) {
// Exclude entries that would be trimmed.
if ($entry->timestamp < CarbonImmutable::now()->subMinutes($period)->getTimestamp()) {
continue;
}
$aggregates = [];

foreach ($entries as $entry) {
foreach ($periods as $period) {
// TODO: add back the comment
if ($entry->timestamp < CarbonImmutable::now()->subMinutes($period)->getTimestamp()) {
continue;
}

yield [
'bucket' => (int) (floor($entry->timestamp / $period) * $period),
$bucket = (int) (floor($entry->timestamp / $period) * $period);

$key = $entry->type.':'.$period.':'.$bucket.':'.$entry->key;

if (! isset($aggregates[$key])) {
$aggregates[$key] = [
'bucket' => $bucket,
'period' => $period,
'type' => $entry->type,
'aggregate' => $aggregateSuffix,
'aggregate' => 'count',
'key' => $entry->key,
'value' => $aggregateSuffix === 'count'
? 1
: $entry->value,
...($aggregateSuffix === 'avg')
? ['count' => 1]
: [],
'value' => 1,
];
} else {
$aggregates[$key]['value']++;
}
}
});

$collapsed = match ($aggregateSuffix) {
'count' => $this->collapseCounts(collect($aggregates)),
'max' => $this->collapseMaxes(collect($aggregates)),
'avg' => $this->collapseAverages(collect($aggregates)),
};
}

return $collapsed;
return collect(array_values($aggregates));

Check failure on line 252 in src/Storage/DatabaseStorage.php

View workflow job for this annotation

GitHub Actions / Static Analysis 

Method Laravel\Pulse\Storage\DatabaseStorage::aggregateCounts() should return Illuminate\Support\Collection<int, array{bucket: int, period: int, type: string, aggregate: string, key: string, value: int, count: int}> but returns Illuminate\Support\Collection<int, array{aggregate: string, bucket: int, key: string, period: int, type: string, value: int}>.
}

protected function collapseCounts(Collection $entries): Collection
/**
* Get the maximum aggregates
*
* @param \Illuminate\Support\Collection<int, \Laravel\Pulse\Entry> $entries
* @param list<int> $periods
* @return \Illuminate\Support\Collection<int, AggregateRow>
*/
protected function aggregateMaximums(Collection $entries, array $periods): Collection
{
return $entries
->groupBy(fn ($value) => implode('-', Arr::only($value, ['bucket', 'period', 'type', 'aggregate', 'key'])))
->map(fn ($values) => [
...$values[0],
'value' => $values->count(),
])
->values();
$aggregates = [];

foreach ($entries as $entry) {
foreach ($periods as $period) {
if ($entry->timestamp < CarbonImmutable::now()->subMinutes($period)->getTimestamp()) {
continue;
}

$bucket = (int) (floor($entry->timestamp / $period) * $period);

$key = $entry->type.':'.$period.':'.$bucket.':'.$entry->key;

if (! isset($aggregates[$key])) {
$aggregates[$key] = [
'bucket' => $bucket,
'period' => $period,
'type' => $entry->type,
'aggregate' => 'max',
'key' => $entry->key,
'value' => $entry->value,
];
} else {
$aggregates[$key]['value'] = max($aggregates[$key]['value'], $entry->value);
}
}
}

return collect(array_values($aggregates));

Check failure on line 291 in src/Storage/DatabaseStorage.php

View workflow job for this annotation

GitHub Actions / Static Analysis 

Method Laravel\Pulse\Storage\DatabaseStorage::aggregateMaximums() should return Illuminate\Support\Collection<int, array{bucket: int, period: int, type: string, aggregate: string, key: string, value: int, count: int}> but returns Illuminate\Support\Collection<int, array{bucket: int, period: int, type: string, aggregate: string, key: string, value: int|null}>.
}

protected function collapseMaxes(Collection $entries): Collection
/**
* Get the average aggregates
*
* @param \Illuminate\Support\Collection<int, \Laravel\Pulse\Entry> $entries
* @param list<int> $periods
* @return \Illuminate\Support\Collection<int, AggregateRow>
*/
protected function aggregateAverages(Collection $entries): Collection

Check failure on line 301 in src/Storage/DatabaseStorage.php

View workflow job for this annotation

GitHub Actions / Static Analysis 

PHPDoc tag @param references unknown parameter: $periods
{
return $entries
->groupBy(fn ($value) => implode('-', Arr::only($value, ['bucket', 'period', 'type', 'aggregate', 'key'])))
->map(fn ($values) => [
...$values[0],
'value' => $values->max('value'),
])
->values();
$aggregates = [];

foreach ($entries as $entry) {
foreach ($this->periods() as $period) {
if ($entry->timestamp < CarbonImmutable::now()->subMinutes($period)->getTimestamp()) {
continue;
}

$bucket = (int) (floor($entry->timestamp / $period) * $period);

$key = $entry->type.':'.$period.':'.$bucket.':'.$entry->key;

if (! isset($aggregates[$key])) {
$aggregates[$key] = [
'bucket' => $bucket,
'period' => $period,
'type' => $entry->type,
'aggregate' => 'avg',
'key' => $entry->key,
'value' => $entry->value,
'count' => 1,
];
} else {
$aggregates[$key]['value'] = ($aggregates[$key]['value'] * $aggregates[$key]['count'] + $entry->value) / ($aggregates[$key]['count'] + 1);
$aggregates[$key]['count']++;
}
}
}

return collect(array_values($aggregates));

Check failure on line 332 in src/Storage/DatabaseStorage.php

View workflow job for this annotation

GitHub Actions / Static Analysis 

Method Laravel\Pulse\Storage\DatabaseStorage::aggregateAverages() should return Illuminate\Support\Collection<int, array{bucket: int, period: int, type: string, aggregate: string, key: string, value: int, count: int}> but returns Illuminate\Support\Collection<int, array{aggregate: string, bucket: int, count: int, key: string, period: mixed, type: string, value: float|int|null}>.
}

protected function collapseAverages(Collection $entries): Collection
protected function periods()

Check failure on line 335 in src/Storage/DatabaseStorage.php

View workflow job for this annotation

GitHub Actions / Static Analysis 

Method Laravel\Pulse\Storage\DatabaseStorage::periods() has no return type specified.
{
return $entries
->groupBy(fn ($value) => implode('-', Arr::only($value, ['bucket', 'period', 'type', 'aggregate', 'key'])))
->map(fn ($values) => [
...$values[0],
'value' => $values->avg('value'),
'count' => $values->count(),
])
->values();
return [
(int) (CarbonInterval::hour()->totalSeconds / 60),
(int) (CarbonInterval::hours(6)->totalSeconds / 60),
(int) (CarbonInterval::hours(24)->totalSeconds / 60),
(int) (CarbonInterval::days(7)->totalSeconds / 60),
];
}

/**
Expand Down

0 comments on commit 945991e

Please sign in to comment.