Repository navigation
Expand file tree
/
Copy pathindex.js
More file actions
271 lines (254 loc) · 9.92 KB
/
Copy pathindex.js
File metadata and controls
271 lines (254 loc) · 9.92 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
const { AsyncLocalStorage } = require("node:async_hooks");
const knex = require("knex");
const util = require("util");
const AppConfig = require("./src/app_config");
const ReportProcessingContext = require("./src/report_processing_context");
const Logger = require("./src/logger");
const Processor = require("./src/processor");
const Queue = require("./src/queue/queue");
const ReportJobQueueMessage = require("./src/queue/report_job_queue_message");
/**
* Gets an array of JSON report objects from the application confing, then runs
* a sequential chain of actions on each report object in the array. Some of the
* actions performed are optional based on the options passed to this function.
*
* @param {object} options an object with options to be used when processing
* all reports.
* @param {boolean} options.csv if true, format report data to CSV
* @param {boolean} options.json if true, format report data to JSON
* @param {string} options.output a string filepath where the analytics data
* will be written to disk after processing.
* @param {boolean} options.publish if true, the analytics data will be written
* to AWS S3 after processing.
* @param {boolean} options.realtime if true, the application will use the
* google analytics realtime data reporting API when requesting data. Otherwise
* the application uses the non-realtime data reporting API.
* @param {boolean} options.slim if true, the application will create a smaller
* data object when formatting the processed data.
* @param {boolean} options.'write-to-database' if true, the application will
* write the processed analytics data to the postgres database.
* @param {string} options.only if set, runs only the report with name
* matching the passed string.
* @param {string} options.frequency if set, runs only the reports with
* frequency matching the passed string.
*/
async function run(options = {}) {
const appConfig = new AppConfig(options);
const context = new ReportProcessingContext(new AsyncLocalStorage());
const reportConfigs = appConfig.filteredReportConfigurations;
const knexInstance = appConfig.shouldWriteToDatabase
? await knex(appConfig.knexConfig)
: undefined;
const processor = Processor.buildAnalyticsProcessor(
appConfig,
Logger.initialize({
agencyName: appConfig.agencyLogName,
scriptName: appConfig.scriptName,
}),
knexInstance,
);
for (const reportConfig of reportConfigs) {
await _processReport(appConfig, context, reportConfig, processor);
}
}
/**
* Creates a new ReportProcessingContext run for the processing of the report.
* Adds data to the context store which can be used by all actions in the
* processor chain (config, logger, reportConfig). Catches any errors that
* occur during processing and logs success or failure. This method does not
* throw so that subsequent reports will run if an error occurs in a previous
* report.
*
* @param {import('./src/app_config')} appConfig the application config
* @param {import('./src/report_processing_context')} context the
* report-specific context for processing.
* @param {object} reportConfig the configuration object for the analytics
* report to process.
* @param {import('./src/processor')} processor an initialized processor
* instance.
* @returns {Promise} resolves when processing completes or has an error.
*/
async function _processReport(appConfig, context, reportConfig, processor) {
return context.run(async () => {
const logger = Logger.initialize({
agencyName: appConfig.agencyLogName,
scriptName: appConfig.scriptName,
reportName: reportConfig.name,
});
context.appConfig = appConfig;
context.logger = logger;
context.reportConfig = reportConfig;
try {
await processor.processChain(context);
logger.info("Processing complete");
} catch (e) {
logger.error("Encountered an error during report processing");
logger.error(util.inspect(e));
}
});
}
/**
* Gets an array of JSON report objects from the application confing, then runs
* a sequential chain of actions on each report object in the array. Some of the
* actions performed are optional based on the options passed to this function.
*
* @param {object} options an object with options to be used when processing
* all reports.
* @param {boolean} options.csv if true, format report data to CSV
* @param {boolean} options.json if true, format report data to JSON
* @param {string} options.output a string filepath where the analytics data
* will be written to disk after processing.
* @param {boolean} options.publish if true, the analytics data will be written
* to AWS S3 after processing.
* @param {boolean} options.realtime if true, the application will use the
* google analytics realtime data reporting API when requesting data. Otherwise
* the application uses the non-realtime data reporting API.
* @param {boolean} options.slim if true, the application will create a smaller
* data object when formatting the processed data.
* @param {boolean} options.'write-to-database' if true, the application will
* write the processed analytics data to the postgres database.
* @param {string} options.only if set, runs only the report with name
* matching the passed string.
* @param {string} options.frequency if set, runs only the reports with
* frequency matching the passed string.
* @param {string} options.agenciesFile if set, run the queue publisher for all
* agencies in the JSON file
*/
async function runQueuePublish(options = {}) {
const agencies = _initAgencies(options.agenciesFile);
const appConfig = new AppConfig(options);
const reportConfigs = appConfig.filteredReportConfigurations;
const appLogger = Logger.initialize({
agencyName: appConfig.agencyLogName,
scriptName: appConfig.scriptName,
});
const knexInstance = await knex(appConfig.knexConfig);
const queueClient = await _initQueueClient(
knexInstance,
appConfig.messageQueueName,
appLogger,
);
for (const agency of agencies) {
for (const reportConfig of reportConfigs) {
process.env.AGENCY_NAME = agency.agencyName;
const reportLogger = Logger.initialize({
agencyName: appConfig.agencyLogName,
scriptName: appConfig.scriptName,
reportName: reportConfig.name,
});
let messageId;
try {
messageId = await queueClient.sendMessage(
new ReportJobQueueMessage({
agencyName: agency.agencyName,
analyticsReportIds: agency.analyticsReportIds,
awsBucketPath: agency.awsBucketPath,
reportOptions: options,
reportConfig,
scriptName: appConfig.scriptName,
}),
);
if (messageId) {
reportLogger.info(
`Created message in queue: ${queueClient.name} with message ID: ${messageId}`,
);
} else {
reportLogger.info(
`Found a duplicate message in queue: ${queueClient.name}`,
);
}
} catch (e) {
// Do nothing so that the remaining messages still process.
}
}
}
try {
await queueClient.stop();
} finally {
appLogger.debug(`Destroying database connection pool`);
knexInstance.destroy();
}
}
function _initAgencies(agencies_file) {
const legacyAgencies = [
{
analyticsReportIds: process.env.ANALYTICS_REPORT_IDS,
agencyName: process.env.AGENCY_NAME,
awsBucketPath: process.env.AWS_BUCKET_PATH,
},
];
if (!agencies_file) {
return legacyAgencies;
}
const agencies = require(agencies_file);
return Array.isArray(agencies) ? agencies : legacyAgencies;
}
async function _initQueueClient(knexInstance, queueName, logger) {
const queueClient = Queue.buildQueue({
knexInstance,
queueName,
messageClass: ReportJobQueueMessage,
logger,
});
await queueClient.start();
return queueClient;
}
/**
* The default job wrapper. Runs the job with no instrumentation. Entry points
* which have an APM agent loaded should pass their own wrapper instead. See
* deploy/consumer.js.
*
* @param {string} name the name of the job being run. Unused by the default
* wrapper.
* @param {Function} job the function which performs the job.
* @returns {Promise} resolves when the job completes.
*/
function _runJobUninstrumented(name, job) {
return job();
}
/**
* @param {object} options an object with options to be used when consuming
* report messages from the queue.
* @param {Function} options.wrapJob a function which receives a job name and
* the function which performs the job, and is responsible for invoking that
* function. Allows an entry point to wrap each job in an APM transaction
* without this module depending on an APM library. Defaults to running the job
* without instrumentation.
* @returns {Promise} when the process ends
*/
async function runQueueConsume({ wrapJob = _runJobUninstrumented } = {}) {
const appConfig = new AppConfig();
const appLogger = Logger.initialize();
const knexInstance = await knex(appConfig.knexConfig);
const queueClient = await _initQueueClient(
knexInstance,
appConfig.messageQueueName,
appLogger,
);
try {
const context = new ReportProcessingContext(new AsyncLocalStorage());
const processor = Processor.buildAnalyticsProcessor(
appConfig,
appLogger,
knexInstance,
);
await queueClient.poll(async (message) => {
return wrapJob("ProcessReport", async () => {
process.env.AGENCY_NAME = message.agencyName;
process.env.ANALYTICS_REPORT_IDS = message.analyticsReportIds;
process.env.AWS_BUCKET_PATH = message.awsBucketPath;
process.env.ANALYTICS_SCRIPT_NAME = message.scriptName;
await _processReport(
new AppConfig(message.options),
context,
message.reportConfig,
processor,
);
});
});
} catch (e) {
appLogger.error("Error polling queue for messages");
appLogger.error(util.inspect(e));
}
}
module.exports = { run, runQueuePublish, runQueueConsume };