17182777590

Committed 11 Aug 2025 04:12PM UTC coverage: 70.85% (+1.0%) from 69.803%

Build # 17182777590

Build Type

push

github

Committed by

web-flow

Commit Message

Make sure some logs that are long are in fact returned in its entirety (#163)

Coverage Stats

64 of 115 branches covered (55.65%)

Branch coverage included in aggregate %.

21 of 22 new or added lines in 1 file covered. (95.45%)

286 of 379 relevant lines covered (75.46%)

50.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.62

/src/createCloudJob.ts

/* eslint no-unused-vars: "off" */
/* eslint @typescript-eslint/no-unused-vars: "error" */
import { Job } from './job'
import * as AWS from 'aws-sdk'
import * as util from 'util'
import * as sleep from 'await-sleep'
import Publisher from './publisher'
import { Runlog } from './RunLog'
import promisify from './promisify'

import * as crypto from 'crypto'

const clusterName = 'violinist-cluster'
const sleepWhilePolling = 5000
// 3 hours (180 mins) hours with the interval above.
const totalRetriesAllowed = (180 * (60 * 1000)) / sleepWhilePolling

const createLogGroup = (taskDefinition) => {
  return util.format('/ecs/%s', taskDefinition)
}

async function getAllLogEvents ({ logs, logGroupName, logStreamName, startTime = 0, startFromHead = true, pageLimit = 1000 }) {
  let nextToken
  const events = []
  const seen = new Set()
  let guard = 0

  while (true) {
    const res = await logs.getLogEvents({
      logGroupName,
      logStreamName,
      startTime,
      startFromHead,
      limit: pageLimit,
      nextToken
    }).promise()

    for (const e of res.events) {
      // We only want unique events, so we use a Set to track seen event IDs.
      // This prevents duplicates in the final events array.
      const hash = crypto.createHash('sha1')
      hash.update(e.message + e.timestamp)
      e.eventId = hash.digest('hex')
      if (!seen.has(e.eventId)) {
        seen.add(e.eventId)
        events.push(e)
      }
    }

    // If the forward token didn't change, there are no more pages.
    if (res.nextForwardToken === nextToken) {
      break
    };
    nextToken = res.nextForwardToken

    // Safety to avoid accidental infinite loops
    if (++guard > 50) {
      throw new Error('Pagination guard tripped')
    }
  }

  return events
}

const createEcsName = (data) => {
  // Should be named like this:
  // PHP version 7.1 => 71
  // PHP version 8.0 => 80
  // ...and so on.
  if (!data.php_version) {
    data.php_version = '7.2'
  }
  return data.php_version.replace('.', '')
}

const createEcsTaskDefinition = (data) => {
  // Should be named like this:
  // violinist-71-composer-1
  // Where 71 means version 7.1, and 1 means composer version 1.
  if (!data.composer_version) {
    data.composer_version = 1
  }
  return util.format('violinist-%s-composer-%s', createEcsName(data), data.composer_version)
}

const createCloudJob = (config, job: Job, gitRev) => {
  return async function runJob (callback) {
    const logData = job.data
    const data = job.data
    const name = createEcsName(data)
    const taskDefinition = createEcsTaskDefinition(data)
    logData.cloud = true
    logData.name = name
    logData.taskDefinition = taskDefinition
    var runLog = new Runlog(logData)
    runLog.log.info('Trying to start cloud job for ' + logData.slug)
    try {
      const awsconfig = {
        accessKeyId: config.accessKeyId,
        secretAccessKey: config.secretAccessKey,
        region: config.region,
        apiVersion: config.apiVersion
      }
      data.violinist_revision = gitRev
      // This log data property is not something we want as ENV. Also, it fails, since it is a boolean.
      delete data.cloud
      // And this also fails since it is a number.
      delete data.queueLength
      // This is also a number, but since we need it, let's convert it to a string.
      if (data.composer_version) {
        data.composer_version = data.composer_version.toString()
      }
      const env = Object.keys(data).map(key => {
        return {
          name: key,
          value: job.data[key] + ''
        }
      })
      env.push({
        name: 'violinist_hostname',
        // We use this to identify the runners are from the cloud.
        value: 'violinist-e2'
      })
      const ecsClient = new AWS.ECS(awsconfig)
      const watchClient = new AWS.CloudWatchLogs(awsconfig)
      const startTime = Date.now()
      const taskData = await ecsClient.runTask({
        cluster: clusterName,
        count: 1,
        launchType: 'FARGATE',
        networkConfiguration: {
          awsvpcConfiguration: {
            assignPublicIp: 'ENABLED',
            subnets: [
              config.subnet
            ]
          }
        },
        overrides: {
          containerOverrides: [
            {
              environment: env,
              name: name
            }
          ]
        },
        taskDefinition: taskDefinition
      }).promise()
      if (!taskData.tasks.length) {
        throw new Error('No valid ARN found')
      }
      const task = taskData.tasks[0]
      const taskArn = task.taskArn
      // Now try to find it again. Until it's stopped, in fact.
      // @todo: Seems we should abstract this a bit, since we are using it twice now.
      let retriesFind = 0
      while (true) {
        retriesFind++
        const foundTask = await ecsClient.describeTasks({
          cluster: clusterName,
          tasks: [taskArn]
        }).promise()
        if (retriesFind > totalRetriesAllowed) {
          throw new Error('Timed out waiting for the job to stop the container. You can try to requeue the project or try again later')
        }
        if (foundTask.tasks && foundTask.tasks.length && foundTask.tasks[0].containers && foundTask.tasks[0].containers.length && foundTask.tasks[0].containers[0].lastStatus) {
          const status = foundTask.tasks[0].containers[0].lastStatus
          if (status === 'STOPPED') {
            break
          }
        }
        await sleep(sleepWhilePolling)
      }
      const arnParts = taskArn.split('/')
      let retries = 0
      let events = []
      while (true) {
        try {
          retries++
          events = await getAllLogEvents({
            logs: watchClient,
            logGroupName: createLogGroup(taskDefinition),
            logStreamName: util.format('ecs/%s/%s', name, arnParts[2])
          })
          // We require the first one to start with "[{" since thats the opening of the JSON stream.
          // But also, the last one should end with "}]"
          if (events.length && events[0].message.startsWith('[{') && events[events.length - 1].message.endsWith('}]')) {
            break
          }
        } catch (logErr) {

        }
        // We are allowed to wait for 3 hours. Thats a very long time, by the way...
        if (retries > totalRetriesAllowed) {
          throw new Error('Timed out waiting for the job to complete and have a log available. You can try to requeue the project or try again later')
        }
        await sleep(sleepWhilePolling)
      }
      runLog.log.info({ eventsLength: events.length }, 'Events length was: ' + events.length)
      const totalTime = Date.now() - startTime
      runLog.log.info({ containerTime: totalTime }, 'Total time was ' + totalTime)
      const stdout = events.map(event => {
        return event.message
      })
      const message = {
        stdout,
        stderr: ''
      }
      const updateData = {
        jobId: data.job_id,
        token: config.token,
        message,
        set_state: 'success'
      }
      const publisher = new Publisher(config)
      const statusData = await promisify(publisher.publish.bind(publisher, updateData))
      runLog.log.info('Job complete request code: ' + statusData.statusCode)
      callback()
    } catch (err) {
      runLog.log.error(err, 'There was an error running a cloud task')
      // Let's create an indication that this did not go so well, did it.
      const message = {
        stdout: [
          JSON.stringify([{
            message: 'There was an error completing the job task. The error message was: ' + err.message,
            type: 'message',
            timestamp: Math.floor(Date.now() / 1000)
          }])
        ],
        stderr: ''
      }
      const updateData = {
        jobId: data.job_id,
        token: config.token,
        message,
        // This field is actually not even used at the moment I think. That's too bad.
        set_state: 'failure'
      }
      const publisher = new Publisher(config)
      const statusData = await promisify(publisher.publish.bind(publisher, updateData))
      runLog.log.info('Job complete request code: ' + statusData.statusCode)
      // We do not care if things go ok, since things are queued so many times anyway.
      callback()
    }
  }
}

export { createCloudJob, createEcsTaskDefinition, createLogGroup }

1	/* eslint no-unused-vars: "off" */
2	/* eslint @typescript-eslint/no-unused-vars: "error" */
3	import { Job } from './job'
4	import * as AWS from 'aws-sdk'	8✔
5	import * as util from 'util'	8✔
6	import * as sleep from 'await-sleep'	8✔
7	import Publisher from './publisher'	8✔
8	import { Runlog } from './RunLog'	8✔
9	import promisify from './promisify'	8✔
10
11	import * as crypto from 'crypto'	8✔
12
13	const clusterName = 'violinist-cluster'	8✔
14	const sleepWhilePolling = 5000	8✔
15	// 3 hours (180 mins) hours with the interval above.
16	const totalRetriesAllowed = (180 * (60 * 1000)) / sleepWhilePolling	8✔
17
18	const createLogGroup = (taskDefinition) => {	8✔
19	return util.format('/ecs/%s', taskDefinition)	46✔
20	}
21
22	async function getAllLogEvents ({ logs, logGroupName, logStreamName, startTime = 0, startFromHead = true, pageLimit = 1000 }) {	48!
23	let nextToken
24	const events = []	24✔
25	const seen = new Set()	24✔
26	let guard = 0	24✔
27
28	while (true) {	48!
29	const res = await logs.getLogEvents({	48✔
30	logGroupName,
31	logStreamName,
32	startTime,
33	startFromHead,
34	limit: pageLimit,
35	nextToken
36	}).promise()
37
38	for (const e of res.events) {	96✔
39	// We only want unique events, so we use a Set to track seen event IDs.
40	// This prevents duplicates in the final events array.
41	const hash = crypto.createHash('sha1')	96✔
42	hash.update(e.message + e.timestamp)	96✔
43	e.eventId = hash.digest('hex')	96✔
44	if (!seen.has(e.eventId)) {	96✔
45	seen.add(e.eventId)	48✔
46	events.push(e)	48✔
47	}
48	}
49
50	// If the forward token didn't change, there are no more pages.
51	if (res.nextForwardToken === nextToken) {	48✔
52	break	24✔
53	};
54	nextToken = res.nextForwardToken	24✔
55
56	// Safety to avoid accidental infinite loops
57	if (++guard > 50) {	24!
NEW 58	throw new Error('Pagination guard tripped')	×
59	}
60	}
61
62	return events	24✔
63	}
64
65	const createEcsName = (data) => {	8✔
66	// Should be named like this:
67	// PHP version 7.1 => 71
68	// PHP version 8.0 => 80
69	// ...and so on.
70	if (!data.php_version) {	74✔
71	data.php_version = '7.2'	4✔
72	}
73	return data.php_version.replace('.', '')	74✔
74	}
75
76	const createEcsTaskDefinition = (data) => {	8✔
77	// Should be named like this:
78	// violinist-71-composer-1
79	// Where 71 means version 7.1, and 1 means composer version 1.
80	if (!data.composer_version) {	48✔
81	data.composer_version = 1	4✔
82	}
83	return util.format('violinist-%s-composer-%s', createEcsName(data), data.composer_version)	48✔
84	}
85
86	const createCloudJob = (config, job: Job, gitRev) => {	8✔
87	return async function runJob (callback) {	26✔
88	const logData = job.data	26✔
89	const data = job.data	26✔
90	const name = createEcsName(data)	26✔
91	const taskDefinition = createEcsTaskDefinition(data)	26✔
92	logData.cloud = true	26✔
93	logData.name = name	26✔
94	logData.taskDefinition = taskDefinition	26✔
95	var runLog = new Runlog(logData)	26✔
96	runLog.log.info('Trying to start cloud job for ' + logData.slug)	26✔
97	try {
98	const awsconfig = {	26✔
99	accessKeyId: config.accessKeyId,
100	secretAccessKey: config.secretAccessKey,
101	region: config.region,
102	apiVersion: config.apiVersion
103	}
104	data.violinist_revision = gitRev	26✔
105	// This log data property is not something we want as ENV. Also, it fails, since it is a boolean.
106	delete data.cloud	26✔
107	// And this also fails since it is a number.
108	delete data.queueLength	26✔
109	// This is also a number, but since we need it, let's convert it to a string.
110	if (data.composer_version) {	26!
111	data.composer_version = data.composer_version.toString()	26✔
112	}
113	const env = Object.keys(data).map(key => {	26✔
114	return {	154✔
115	name: key,
116	value: job.data[key] + ''
117	}
118	})
119	env.push({	26✔
120	name: 'violinist_hostname',
121	// We use this to identify the runners are from the cloud.
122	value: 'violinist-e2'
123	})
124	const ecsClient = new AWS.ECS(awsconfig)	26✔
125	const watchClient = new AWS.CloudWatchLogs(awsconfig)	24✔
126	const startTime = Date.now()	24✔
127	const taskData = await ecsClient.runTask({	24✔
128	cluster: clusterName,
129	count: 1,
130	launchType: 'FARGATE',
131	networkConfiguration: {
132	awsvpcConfiguration: {
133	assignPublicIp: 'ENABLED',
134	subnets: [
135	config.subnet
136	]
137	}
138	},
139	overrides: {
140	containerOverrides: [
141	{
142	environment: env,
143	name: name
144	}
145	]
146	},
147	taskDefinition: taskDefinition
148	}).promise()
149	if (!taskData.tasks.length) {	24!
150	throw new Error('No valid ARN found')	×
151	}
152	const task = taskData.tasks[0]	24✔
153	const taskArn = task.taskArn	24✔
154	// Now try to find it again. Until it's stopped, in fact.
155	// @todo: Seems we should abstract this a bit, since we are using it twice now.
156	let retriesFind = 0	24✔
157	while (true) {	24!
158	retriesFind++	24✔
159	const foundTask = await ecsClient.describeTasks({	24✔
160	cluster: clusterName,
161	tasks: [taskArn]
162	}).promise()
163	if (retriesFind > totalRetriesAllowed) {	24!
164	throw new Error('Timed out waiting for the job to stop the container. You can try to requeue the project or try again later')	×
165	}
166	if (foundTask.tasks && foundTask.tasks.length && foundTask.tasks[0].containers && foundTask.tasks[0].containers.length && foundTask.tasks[0].containers[0].lastStatus) {	24!
167	const status = foundTask.tasks[0].containers[0].lastStatus	24✔
168	if (status === 'STOPPED') {	24!
169	break	24✔
170	}
171	}
172	await sleep(sleepWhilePolling)	×
173	}
174	const arnParts = taskArn.split('/')	24✔
175	let retries = 0	24✔
176	let events = []	24✔
177	while (true) {	24!
178	try {
179	retries++	24✔
180	events = await getAllLogEvents({	24✔
181	logs: watchClient,
182	logGroupName: createLogGroup(taskDefinition),
183	logStreamName: util.format('ecs/%s/%s', name, arnParts[2])
184	})
185	// We require the first one to start with "[{" since thats the opening of the JSON stream.
186	// But also, the last one should end with "}]"
187	if (events.length && events[0].message.startsWith('[{') && events[events.length - 1].message.endsWith('}]')) {	24!
188	break	24✔
189	}
190	} catch (logErr) {
191
192	}
193	// We are allowed to wait for 3 hours. Thats a very long time, by the way...
194	if (retries > totalRetriesAllowed) {	×
195	throw new Error('Timed out waiting for the job to complete and have a log available. You can try to requeue the project or try again later')	×
196	}
197	await sleep(sleepWhilePolling)	×
198	}
199	runLog.log.info({ eventsLength: events.length }, 'Events length was: ' + events.length)	24✔
200	const totalTime = Date.now() - startTime	24✔
201	runLog.log.info({ containerTime: totalTime }, 'Total time was ' + totalTime)	24✔
202	const stdout = events.map(event => {	24✔
203	return event.message	48✔
204	})
205	const message = {	24✔
206	stdout,
207	stderr: ''
208	}
209	const updateData = {	24✔
210	jobId: data.job_id,
211	token: config.token,
212	message,
213	set_state: 'success'
214	}
215	const publisher = new Publisher(config)	24✔
216	const statusData = await promisify(publisher.publish.bind(publisher, updateData))	24✔
217	runLog.log.info('Job complete request code: ' + statusData.statusCode)	24✔
218	callback()	24✔
219	} catch (err) {
220	runLog.log.error(err, 'There was an error running a cloud task')	2✔
221	// Let's create an indication that this did not go so well, did it.
222	const message = {	2✔
223	stdout: [
224	JSON.stringify([{
225	message: 'There was an error completing the job task. The error message was: ' + err.message,
226	type: 'message',
227	timestamp: Math.floor(Date.now() / 1000)
228	}])
229	],
230	stderr: ''
231	}
232	const updateData = {	2✔
233	jobId: data.job_id,
234	token: config.token,
235	message,
236	// This field is actually not even used at the moment I think. That's too bad.
237	set_state: 'failure'
238	}
239	const publisher = new Publisher(config)	2✔
240	const statusData = await promisify(publisher.publish.bind(publisher, updateData))	2✔
241	runLog.log.info('Job complete request code: ' + statusData.statusCode)	2✔
242	// We do not care if things go ok, since things are queued so many times anyway.
243	callback()	2✔
244	}
245	}
246	}
247
248	export { createCloudJob, createEcsTaskDefinition, createLogGroup }	8✔

violinist-dev / queue-starter / 17182777590

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous