Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions src/core/public/pulse/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,13 +45,13 @@ const logger = {
export class PulseService {
private retriableErrors = 0;
private readonly channels: Map<string, PulseChannel>;
private readonly instructions: Map<string, Subject<any>> = new Map();
private readonly instructions$: Map<string, Subject<any>> = new Map();

constructor() {
this.channels = new Map(
channelNames.map((id): [string, PulseChannel] => {
const instructions$ = new Subject<PulseInstruction>();
this.instructions.set(id, instructions$);
this.instructions$.set(id, instructions$);
const channel = new PulseChannel({ id, instructions$, logger });
return [channel.id, channel];
})
Expand Down Expand Up @@ -127,8 +127,8 @@ export class PulseService {

const responseBody: InstructionsResponse = await response.json();

responseBody.channels.forEach((channel: PulseChannel) => {
const instructions$ = this.instructions.get(channel.id);
responseBody.channels.forEach(channel => {
const instructions$ = this.instructions$.get(channel.id);
if (!instructions$) {
throw new Error(
`Channel (${channel.id}) from service has no corresponding channel handler in client`
Expand Down
4 changes: 2 additions & 2 deletions src/core/server/pulse/channel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,12 @@ export class PulseChannel<Payload = any, Rec = Payload> {
private readonly collector: any;

constructor(private readonly config: ChannelConfig) {
const Collector: PulseCollectorContructor = require(`${__dirname}/collectors/${this.id}`)
const Collector: PulseCollectorConstructor = require(`${__dirname}/collectors/${this.id}`)
.Collector;
this.collector = new Collector(this.config.logger);
}

public async setup(setupContext: ChannelSetupContexxt) {
public async setup(setupContext: ChannelSetupContext) {
return this.collector.setup(setupContext);
}

Expand Down
2 changes: 1 addition & 1 deletion src/core/server/pulse/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ interface ChannelResponse {
instructions: PulseInstruction[];
}

interface InstructionsResponse {
export interface InstructionsResponse {
channels: ChannelResponse[];
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,29 +9,29 @@ import { CheckContext } from '../../types';

export async function check(es: IScopedClusterClient, context: CheckContext) {
const { deploymentId } = context;
// const response = await es.callAsInternalUser('search', {
// index: 'pulse-poc-raw*',
// size: 0,
// allow_no_indices: true,
// ignore_unavailable: true,
// body: {
// query: {
// term: {
// deployment_id: {
// value: deploymentId,
// },
// },
// },
// },
// });
const response = await es.callAsInternalUser('search', {
index: 'pulse-poc-raw*',
size: 0,
allow_no_indices: true,
ignore_unavailable: true,
body: {
query: {
term: {
deployment_id: {
value: deploymentId,
},
},
},
},
});

// if (response.hits.total.value > 0) {
// return undefined;
// } else {
return {
owner: 'core',
id: 'pulse_telemetry',
value: 'try_again',
};
// }
if (response.hits.total.value > 0) {
return undefined;
} else {
return {
owner: 'core',
id: 'pulse_telemetry',
value: 'try_again',
};
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,12 @@
*/

import { IScopedClusterClient } from 'src/core/server';
import { CheckContext } from '../../types';

export async function check(es: IScopedClusterClient, deploymentId: string) {
export async function check(es: IScopedClusterClient, { deploymentId, indexName }: CheckContext) {
// TODO: modify the search query for full text search
const response = await es.callAsInternalUser('search', {
index: 'pulse-poc-raw-errors',
index: indexName,
size: 100,
allow_no_indices: true,
ignore_unavailable: true,
Expand Down
6 changes: 4 additions & 2 deletions x-pack/plugins/pulse_poc/server/plugin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -134,8 +134,10 @@ export class PulsePocPlugin {
const { deploymentId } = request.params;
const es = context.core.elasticsearch.adminClient;
const allChannelCheckResults = this.channels.map(async channel => {
// const indexName = `pulse-poc-raw-${channel.id}`;
const channelChecks = channel.checks.map(check => check.check(es, deploymentId));
const indexName = `pulse-poc-raw-${channel.id}`;
const channelChecks = channel.checks.map(check =>
check.check(es, { deploymentId, indexName })
);
const checkResults = await Promise.all(channelChecks);
const instructions = checkResults.filter((value: any) => Boolean(value));
return {
Expand Down