enhance: improve federation chart

This commit is contained in:
syuilo 2022-02-10 17:45:12 +09:00
parent 7ba5512a65
commit 0afebcfd9e
25 changed files with 199 additions and 64 deletions

View File

@ -0,0 +1,31 @@
const { MigrationInterface, QueryRunner } = require("typeorm");
module.exports = class chartV151644481657998 {
name = 'chartV151644481657998'
async up(queryRunner) {
await queryRunner.query(`ALTER TABLE "__chart__federation" DROP COLUMN "___instance_total"`);
await queryRunner.query(`ALTER TABLE "__chart__federation" DROP COLUMN "___instance_inc"`);
await queryRunner.query(`ALTER TABLE "__chart__federation" DROP COLUMN "___instance_dec"`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" DROP COLUMN "___instance_total"`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" DROP COLUMN "___instance_inc"`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" DROP COLUMN "___instance_dec"`);
await queryRunner.query(`ALTER TABLE "__chart__federation" ADD "___sub" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart__federation" ADD "___pub" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" ADD "___sub" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" ADD "___pub" smallint NOT NULL DEFAULT '0'`);
}
async down(queryRunner) {
await queryRunner.query(`ALTER TABLE "__chart_day__federation" DROP COLUMN "___pub"`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" DROP COLUMN "___sub"`);
await queryRunner.query(`ALTER TABLE "__chart__federation" DROP COLUMN "___pub"`);
await queryRunner.query(`ALTER TABLE "__chart__federation" DROP COLUMN "___sub"`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" ADD "___instance_dec" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" ADD "___instance_inc" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart_day__federation" ADD "___instance_total" integer NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart__federation" ADD "___instance_dec" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart__federation" ADD "___instance_inc" smallint NOT NULL DEFAULT '0'`);
await queryRunner.query(`ALTER TABLE "__chart__federation" ADD "___instance_total" integer NOT NULL DEFAULT '0'`);
}
}

View File

@ -258,6 +258,11 @@ export default function() {
processDb(dbQueue);
processObjectStorage(objectStorageQueue);
systemQueue.add('tickCharts', {
}, {
repeat: { cron: '55 * * * *' },
});
systemQueue.add('resyncCharts', {
}, {
repeat: { cron: '0 0 * * *' },

View File

@ -1,8 +1,10 @@
import * as Bull from 'bull';
import { tickCharts } from './tick-charts';
import { resyncCharts } from './resync-charts';
import { cleanCharts } from './clean-charts';
const jobs = {
tickCharts,
resyncCharts,
cleanCharts,
} as Record<string, Bull.ProcessCallbackFunction<Record<string, unknown>> | Bull.ProcessPromiseFunction<Record<string, unknown>>>;

View File

@ -0,0 +1,28 @@
import * as Bull from 'bull';
import { queueLogger } from '../../logger';
import { activeUsersChart, driveChart, federationChart, hashtagChart, instanceChart, notesChart, perUserDriveChart, perUserFollowingChart, perUserNotesChart, perUserReactionsChart, usersChart, apRequestChart } from '@/services/chart/index';
const logger = queueLogger.createSubLogger('tick-charts');
export async function tickCharts(job: Bull.Job<Record<string, unknown>>, done: any): Promise<void> {
logger.info(`Tick charts...`);
await Promise.all([
federationChart.tick(false),
notesChart.tick(false),
usersChart.tick(false),
activeUsersChart.tick(false),
instanceChart.tick(false),
perUserNotesChart.tick(false),
driveChart.tick(false),
perUserReactionsChart.tick(false),
hashtagChart.tick(false),
perUserFollowingChart.tick(false),
perUserDriveChart.tick(false),
apRequestChart.tick(false),
]);
logger.succ(`All charts successfully ticked.`);
done();
}

View File

@ -1,22 +0,0 @@
import define from '../../define';
import { driveChart, notesChart, usersChart } from '@/services/chart/index';
import { insertModerationLog } from '@/services/insert-moderation-log';
export const meta = {
tags: ['admin'],
requireCredential: true,
requireModerator: true,
} as const;
// eslint-disable-next-line import/no-default-export
export default define(meta, async (ps, me) => {
insertModerationLog(me, 'chartResync');
driveChart.resync();
notesChart.resync();
usersChart.resync();
// TODO: ユーザーごとのチャートもキューに入れて更新する
// TODO: インスタンスごとのチャートもキューに入れて更新する
});

View File

@ -18,7 +18,12 @@ export default class ActiveUsersChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -12,7 +12,12 @@ export default class ApRequestChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -15,7 +15,12 @@ export default class DriveChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -3,12 +3,11 @@ import Chart from '../../core';
export const name = 'federation';
export const schema = {
'instance.total': { accumulate: true },
'instance.inc': { range: 'small' },
'instance.dec': { range: 'small' },
'deliveredInstances': { uniqueIncrement: true, range: 'small' },
'inboxInstances': { uniqueIncrement: true, range: 'small' },
'stalled': { uniqueIncrement: true, range: 'small' },
'sub': { accumulate: true, range: 'small' },
'pub': { accumulate: true, range: 'small' },
} as const;
export const entity = Chart.schemaToEntity(name, schema);

View File

@ -1,6 +1,6 @@
import autobind from 'autobind-decorator';
import Chart, { KVs } from '../core';
import { Instances } from '@/models/index';
import { Followings } from '@/models/index';
import { name, schema } from './entities/federation';
/**
@ -13,23 +13,30 @@ export default class FederationChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
const [total] = await Promise.all([
Instances.count({}),
]);
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {
'instance.total': total,
};
}
@autobind
public async update(isAdditional: boolean): Promise<void> {
await this.commit({
'instance.total': isAdditional ? 1 : -1,
'instance.inc': isAdditional ? 1 : 0,
'instance.dec': isAdditional ? 0 : 1,
});
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
const [sub, pub] = await Promise.all([
Followings.createQueryBuilder('following')
.select('COUNT(DISTINCT following.followeeHost)')
.where('following.followeeHost IS NOT NULL')
.getRawOne()
.then(x => parseInt(x.count, 10)),
Followings.createQueryBuilder('following')
.select('COUNT(DISTINCT following.followerHost)')
.where('following.followerHost IS NOT NULL')
.getRawOne()
.then(x => parseInt(x.count, 10)),
]);
return {
'sub': sub,
'pub': pub,
};
}
@autobind

View File

@ -14,7 +14,12 @@ export default class HashtagChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -16,7 +16,7 @@ export default class InstanceChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(group: string): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(group: string): Promise<Partial<KVs<typeof schema>>> {
const [
notesCount,
usersCount,
@ -42,6 +42,11 @@ export default class InstanceChart extends Chart<typeof schema> {
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async requestReceived(host: string): Promise<void> {
await this.commit({

View File

@ -15,7 +15,7 @@ export default class NotesChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
const [localCount, remoteCount] = await Promise.all([
Notes.count({ userHost: null }),
Notes.count({ userHost: Not(IsNull()) }),
@ -27,6 +27,11 @@ export default class NotesChart extends Chart<typeof schema> {
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async update(note: Note, isAdditional: boolean): Promise<void> {
const prefix = note.userHost === null ? 'local' : 'remote';

View File

@ -14,7 +14,7 @@ export default class PerUserDriveChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(group: string): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(group: string): Promise<Partial<KVs<typeof schema>>> {
const [count, size] = await Promise.all([
DriveFiles.count({ userId: group }),
DriveFiles.calcDriveUsageOf(group),
@ -26,6 +26,11 @@ export default class PerUserDriveChart extends Chart<typeof schema> {
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async update(file: DriveFile, isAdditional: boolean): Promise<void> {
const fileSizeKb = file.size / 1000;

View File

@ -15,7 +15,7 @@ export default class PerUserFollowingChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(group: string): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(group: string): Promise<Partial<KVs<typeof schema>>> {
const [
localFollowingsCount,
localFollowersCount,
@ -36,6 +36,11 @@ export default class PerUserFollowingChart extends Chart<typeof schema> {
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async update(follower: { id: User['id']; host: User['host']; }, followee: { id: User['id']; host: User['host']; }, isFollow: boolean): Promise<void> {
const prefixFollower = Users.isLocalUser(follower) ? 'local' : 'remote';

View File

@ -15,7 +15,7 @@ export default class PerUserNotesChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(group: string): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(group: string): Promise<Partial<KVs<typeof schema>>> {
const [count] = await Promise.all([
Notes.count({ userId: group }),
]);
@ -25,6 +25,11 @@ export default class PerUserNotesChart extends Chart<typeof schema> {
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async update(user: { id: User['id'] }, note: Note, isAdditional: boolean): Promise<void> {
await this.commit({

View File

@ -15,7 +15,12 @@ export default class PerUserReactionsChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(group: string): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(group: string): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -14,12 +14,17 @@ export default class TestGroupedChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(group: string): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(group: string): Promise<Partial<KVs<typeof schema>>> {
return {
'foo.total': this.total[group],
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async increment(group: string): Promise<void> {
if (this.total[group] == null) this.total[group] = 0;

View File

@ -12,7 +12,12 @@ export default class TestIntersectionChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -12,7 +12,12 @@ export default class TestUniqueChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}

View File

@ -14,12 +14,17 @@ export default class TestChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
return {
'foo.total': this.total,
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async increment(): Promise<void> {
this.total++;

View File

@ -15,7 +15,7 @@ export default class UsersChart extends Chart<typeof schema> {
}
@autobind
protected async queryCurrentState(): Promise<Partial<KVs<typeof schema>>> {
protected async tickMajor(): Promise<Partial<KVs<typeof schema>>> {
const [localCount, remoteCount] = await Promise.all([
Users.count({ host: null }),
Users.count({ host: Not(IsNull()) }),
@ -27,6 +27,11 @@ export default class UsersChart extends Chart<typeof schema> {
};
}
@autobind
protected async tickMinor(): Promise<Partial<KVs<typeof schema>>> {
return {};
}
@autobind
public async update(user: { id: User['id'], host: User['host'] }, isAdditional: boolean): Promise<void> {
const prefix = Users.isLocalUser(user) ? 'local' : 'remote';

View File

@ -81,7 +81,15 @@ export default abstract class Chart<T extends Schema> {
protected repositoryForHour: Repository<RawRecord<T>>;
protected repositoryForDay: Repository<RawRecord<T>>;
protected abstract queryCurrentState(group: string | null): Promise<Partial<KVs<T>>>;
/**
* 1(CASCADE削除などアプリケーション側で感知できない変動によるズレの修正用)
*/
protected abstract tickMajor(group: string | null): Promise<Partial<KVs<T>>>;
/**
* 1
*/
protected abstract tickMinor(group: string | null): Promise<Partial<KVs<T>>>;
@autobind
private static convertSchemaToColumnDefinitions(schema: Schema): Record<string, { type: string; array?: boolean; default?: any; }> {
@ -445,8 +453,8 @@ export default abstract class Chart<T extends Schema> {
}
@autobind
public async resync(group: string | null = null): Promise<void> {
const data = await this.queryCurrentState(group);
public async tick(major: boolean, group: string | null = null): Promise<void> {
const data = major ? await this.tickMajor(group) : await this.tickMinor(group);
const columns = {} as Record<string, number>;
for (const [k, v] of Object.entries(data)) {
@ -480,6 +488,11 @@ export default abstract class Chart<T extends Schema> {
update(logHour, logDay));
}
@autobind
public resync(group: string | null = null): Promise<void> {
return this.tick(true, group);
}
@autobind
public async clean(): Promise<void> {
const current = dateUTC(Chart.getCurrentDate());

View File

@ -1,6 +1,5 @@
import { Instance } from '@/models/entities/instance';
import { Instances } from '@/models/index';
import { federationChart } from '@/services/chart/index';
import { genId } from '@/misc/gen-id';
import { toPuny } from '@/misc/convert-host';
import { Cache } from '@/misc/cache';
@ -23,8 +22,6 @@ export async function registerOrFetchInstanceDoc(host: string): Promise<Instance
lastCommunicatedAt: new Date(),
}).then(x => Instances.findOneOrFail(x.identifiers[0]));
federationChart.update(true);
cache.set(host, i);
return i;
} else {

View File

@ -370,14 +370,14 @@ export default defineComponent({
const raw = await os.api('charts/federation', { limit: props.limit, span: props.span });
return {
series: [{
name: 'Total',
name: 'Sub',
type: 'area',
data: format(raw.instance.total),
color: '#888888',
data: format(raw.sub),
color: colors.orange,
}, {
name: 'Inc/Dec',
name: 'Pub',
type: 'area',
data: format(sum(raw.instance.inc, negate(raw.instance.dec))),
data: format(raw.pub),
color: colors.purple,
}, {
name: 'Received',