conductor-s3.test.js (7331B)
1 // test/conductor-s3.test.js - the conductor running on object storage 2 3 import test, { before, after } from 'node:test'; 4 import assert from 'node:assert/strict'; 5 import fs from 'node:fs/promises'; 6 import { startHarness, data } from './helpers/harness.js'; 7 import { startMinio, s3Available } from './helpers/minio.js'; 8 import { createStorage, keys } from '../src/lib/storage/index.js'; 9 10 const opts = { skip: s3Available() ? false : 'docker is not available' }; 11 12 let minio = null; 13 14 before(async () => { 15 if (s3Available()) minio = await startMinio(); 16 }, { timeout: 240000 }); 17 18 after(async () => { 19 if (minio) await minio.stop(); 20 }); 21 22 async function withS3Conductor(options, fn) { 23 const bucket = await minio.bucket(); 24 const s3 = minio.storageConfig(bucket); 25 const h = await startHarness({ visibility: 'public', s3, ...options }); 26 const store = createStorage({ storage: { driver: 's3', s3 } }); 27 28 try { 29 return await fn(h, { store, bucket }); 30 } finally { 31 await h.stop(); 32 } 33 } 34 35 const octet = (h) => ({ ...h.auth, 'content-type': 'application/octet-stream' }); 36 37 async function drain(stream) { 38 let out = ''; 39 for await (const chunk of stream) out += chunk; 40 return out; 41 } 42 43 async function taskWithOutput(h, { log = 'building\ndone\n', artifact = 'artifact bytes' } = {}) { 44 const triggered = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' }); 45 const jobId = data(triggered).job_id; 46 await h.waitForJob(jobId); 47 48 const claimed = await h.claim({ arches: ['x86_64'] }); 49 assert.equal(claimed.statusCode, 200, claimed.body); 50 const task = data(claimed).task; 51 52 await h.app.inject({ 53 method: 'POST', 54 url: `/api/v1/tasks/${encodeURIComponent(task.id)}/log`, 55 headers: octet(h), 56 payload: Buffer.from(log), 57 }); 58 59 if (artifact !== null) { 60 const upload = await h.app.inject({ 61 method: 'POST', 62 url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`, 63 headers: { ...octet(h), 'x-artifact-path': 'dist/app bin+v1.tar.gz' }, 64 payload: Buffer.from(artifact), 65 }); 66 assert.equal(upload.statusCode, 200, upload.body); 67 } 68 69 return { jobId, task }; 70 } 71 72 function complete(h, taskId) { 73 return h.app.inject({ 74 method: 'POST', 75 url: `/api/v1/tasks/${encodeURIComponent(taskId)}/complete`, 76 headers: { ...h.auth, 'content-type': 'application/json' }, 77 payload: JSON.stringify({ state: 'success', success: true, exit_code: 0 }), 78 }); 79 } 80 81 function readLog(h, taskId, query = '') { 82 return h.app.inject({ method: 'GET', url: `/api/v1/tasks/${encodeURIComponent(taskId)}/log${query}` }); 83 } 84 85 test('the conductor reports object storage when configured for it', opts, async () => { 86 await withS3Conductor({}, async (h) => { 87 const health = await h.app.inject({ method: 'GET', url: '/health' }); 88 assert.equal(data(health).storage, 's3'); 89 }); 90 }); 91 92 test('an artifact uploaded by a worker really lands in the bucket', opts, async () => { 93 await withS3Conductor({}, async (h, { store }) => { 94 const { jobId, task } = await taskWithOutput(h); 95 96 const listed = await h.app.inject({ 97 method: 'GET', url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`, 98 }); 99 const artifact = data(listed)[0]; 100 assert.equal(artifact.path, 'dist/app bin+v1.tar.gz'); 101 102 const key = keys.artifact(jobId, task.id, artifact.path); 103 const stored = await store.get(key); 104 assert.equal(await drain(stored.stream), 'artifact bytes'); 105 assert.equal((await store.head(key)).size, artifact.size); 106 }); 107 }); 108 109 test('an artifact download is handed off with a redirect that works', opts, async () => { 110 await withS3Conductor({}, async (h) => { 111 const { jobId, task } = await taskWithOutput(h); 112 const listed = await h.app.inject({ 113 method: 'GET', url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`, 114 }); 115 const artifact = data(listed)[0]; 116 117 const res = await h.app.inject({ 118 method: 'GET', 119 url: `/api/v1/projects/demo/jobs/${jobId}/tasks/${encodeURIComponent(task.id)}` 120 + `/artifacts/${artifact.id}`, 121 }); 122 assert.equal(res.statusCode, 302, res.body); 123 assert.match(res.headers.location, /X-Amz-Signature=/); 124 125 const followed = await fetch(res.headers.location); 126 assert.equal(followed.status, 200); 127 assert.equal(await followed.text(), 'artifact bytes'); 128 }); 129 }); 130 131 test('a finished log is moved to the bucket and stays readable', opts, async () => { 132 await withS3Conductor({}, async (h, { store }) => { 133 const { jobId, task } = await taskWithOutput(h, { log: 'compiling\nlinking\n' }); 134 135 const live = await readLog(h, task.id); 136 assert.equal(data(live).data, 'compiling\nlinking\n'); 137 138 assert.equal((await complete(h, task.id)).statusCode, 200); 139 140 const archived = await readLog(h, task.id); 141 assert.equal(data(archived).data, 'compiling\nlinking\n'); 142 143 const row = await h.services.db.get('SELECT log_key FROM tasks WHERE id = {id}', { id: task.id }); 144 assert.equal(row.log_key, keys.log(jobId, task.id)); 145 146 const stored = await store.get(row.log_key); 147 assert.equal(await drain(stored.stream), 'compiling\nlinking\n'); 148 }); 149 }); 150 151 test('the local spool is released once the log is archived', opts, async () => { 152 await withS3Conductor({}, async (h) => { 153 const { jobId, task } = await taskWithOutput(h, { log: 'output\n', artifact: null }); 154 155 const spool = h.services.logs.spoolPath(jobId, task.id); 156 assert.ok(await fs.stat(spool).catch(() => null), 'the spool should exist while running'); 157 158 await complete(h, task.id); 159 160 assert.equal(await fs.stat(spool).catch(() => null), null, 'the spool should be gone'); 161 }); 162 }); 163 164 test('an archived log still serves incremental ranges', opts, async () => { 165 await withS3Conductor({}, async (h) => { 166 const { task } = await taskWithOutput(h, { log: 'first line\nsecond line\n', artifact: null }); 167 await complete(h, task.id); 168 169 const tail = await readLog(h, task.id, '?offset=11'); 170 assert.equal(data(tail).data, 'second line\n'); 171 }); 172 }); 173 174 test('a private job in a bucket is not readable by a stranger', opts, async () => { 175 await withS3Conductor({ visibility: 'private' }, async (h) => { 176 const { jobId, task } = await taskWithOutput(h); 177 178 const detail = await h.app.inject({ method: 'GET', url: `/api/v1/jobs/${jobId}` }); 179 assert.equal(detail.statusCode, 404); 180 181 const log = await readLog(h, task.id); 182 assert.equal(log.statusCode, 404); 183 }); 184 }); 185 186 test('a large artifact streams through to the bucket', opts, async () => { 187 await withS3Conductor({}, async (h, { store }) => { 188 const triggered = await h.trigger({ sha: h.sha, ref: 'refs/heads/main' }); 189 const jobId = data(triggered).job_id; 190 await h.waitForJob(jobId); 191 192 const claimed = await h.claim({ arches: ['x86_64'] }); 193 const task = data(claimed).task; 194 const payload = Buffer.alloc(3 * 1024 * 1024, 0x41); 195 196 const upload = await h.app.inject({ 197 method: 'POST', 198 url: `/api/v1/tasks/${encodeURIComponent(task.id)}/artifacts`, 199 headers: { ...octet(h), 'x-artifact-path': 'dist/big.bin' }, 200 payload, 201 }); 202 assert.equal(upload.statusCode, 200, upload.body); 203 assert.equal(data(upload).size, payload.length); 204 205 const stored = await store.head(keys.artifact(jobId, task.id, 'dist/big.bin')); 206 assert.equal(stored.size, payload.length); 207 }); 208 });