conductor-s3.test.js (9581B)
1 // test/conductor-s3.test.js - the conductor running on object storage 2 // 3 // The storage module is covered on its own elsewhere. What is checked here 4 // is the conductor configured to use it: that artifacts actually land in 5 // the bucket, that a finished log is moved there and remains readable, and 6 // that a download is handed off with a redirect instead of being proxied. 7 // 8 // That handoff only happens on S3, since the local backend has no URL to 9 // give out, so none of it is exercised by the default configuration. 10 11 import test, { before, after } from 'node:test'; 12 import assert from 'node:assert/strict'; 13 import fs from 'node:fs/promises'; 14 import { startHarness } from './helpers/harness.js'; 15 import { startMinio, s3Available } from './helpers/minio.js'; 16 import { createStorage, keys } from '../src/lib/storage/index.js'; 17 18 const opts = { skip: s3Available() ? false : 'docker is not available' }; 19 20 let minio = null; 21 22 before(async () => { 23 if (s3Available()) minio = await startMinio(); 24 }, { timeout: 240000 }); 25 26 after(async () => { 27 if (minio) await minio.stop(); 28 }); 29 30 // A conductor whose storage is a fresh bucket, plus a handle on that 31 // bucket so a test can look at what actually landed in it. 32 async function withS3Conductor(options, fn) { 33 const bucket = await minio.bucket(); 34 const s3 = minio.storageConfig(bucket); 35 const h = await startHarness({ visibility: 'public', s3, ...options }); 36 const store = createStorage({ storage: { driver: 's3', s3 } }); 37 38 try { 39 return await fn(h, { store, bucket }); 40 } finally { 41 await h.stop(); 42 } 43 } 44 45 const octet = (h) => ({ ...h.auth, 'content-type': 'application/octet-stream' }); 46 47 async function drain(stream) { 48 let out = ''; 49 for await (const chunk of stream) out += chunk; 50 return out; 51 } 52 53 // Runs a job far enough to have a log and an artifact. 54 async function jobWithOutput(h, { log = 'building\ndone\n', artifact = 'artifact bytes' } = {}) { 55 await h.trigger({ sha: h.sha, ref: 'refs/heads/main' }); 56 const job = (await h.poll({ arches: 'x86_64' })).json().job; 57 58 await h.app.inject({ 59 method: 'POST', 60 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/log`, 61 headers: octet(h), 62 payload: Buffer.from(log), 63 }); 64 65 if (artifact !== null) { 66 await h.app.inject({ 67 method: 'POST', 68 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`, 69 headers: { ...octet(h), 'x-artifact-path': 'dist/app bin+v1.tar.gz' }, 70 payload: Buffer.from(artifact), 71 }); 72 } 73 74 return job; 75 } 76 77 test('the conductor reports object storage when configured for it', opts, async () => { 78 await withS3Conductor({}, async (h) => { 79 const health = (await h.app.inject({ method: 'GET', url: '/health' })).json(); 80 assert.equal(health.storage, 's3'); 81 }); 82 }); 83 84 test('an artifact uploaded by a worker really lands in the bucket', opts, async () => { 85 await withS3Conductor({}, async (h, { store }) => { 86 const job = await jobWithOutput(h); 87 88 const detail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` }); 89 const artifact = detail.json().artifacts[0]; 90 assert.equal(artifact.path, 'dist/app bin+v1.tar.gz'); 91 92 // Read it straight out of the bucket, not through the conductor. 93 const key = keys.artifact(job.run_id, job.id, artifact.path); 94 const stored = await store.get(key); 95 assert.equal(await drain(stored.stream), 'artifact bytes'); 96 assert.equal((await store.head(key)).size, artifact.size); 97 }); 98 }); 99 100 test('an artifact download is handed off with a redirect that works', opts, async () => { 101 await withS3Conductor({}, async (h) => { 102 const job = await jobWithOutput(h); 103 const detail = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` }); 104 const artifact = detail.json().artifacts[0]; 105 106 const res = await h.app.inject({ method: 'GET', url: `/api/artifacts/${artifact.id}` }); 107 108 // With local storage this streams a 200; on S3 the client is sent to 109 // the store so the conductor does not carry the bytes. 110 assert.equal(res.statusCode, 302); 111 const location = res.headers.location; 112 assert.match(location, /X-Amz-Signature=/); 113 114 // The presigned URL has to be usable as given, with no credentials. 115 const followed = await fetch(location); 116 assert.equal(followed.status, 200); 117 assert.equal(await followed.text(), 'artifact bytes'); 118 }); 119 }); 120 121 test('a finished log is moved to the bucket and stays readable', opts, async () => { 122 await withS3Conductor({}, async (h, { store }) => { 123 const job = await jobWithOutput(h, { log: 'compiling\nlinking\n' }); 124 125 // While running it is served from the local spool. 126 const live = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log` }); 127 assert.equal(live.headers['x-log-complete'], 'false'); 128 assert.equal(live.body, 'compiling\nlinking\n'); 129 130 await h.app.inject({ 131 method: 'POST', 132 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`, 133 headers: { ...h.auth, 'content-type': 'application/json' }, 134 payload: JSON.stringify({ success: true, exit_code: 0 }), 135 }); 136 137 // Afterwards it comes from the bucket, and reads the same. 138 const archived = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log` }); 139 assert.equal(archived.headers['x-log-complete'], 'true'); 140 assert.equal(archived.body, 'compiling\nlinking\n'); 141 142 const stored = await store.get(keys.log(job.run_id, job.id)); 143 assert.equal(await drain(stored.stream), 'compiling\nlinking\n'); 144 }); 145 }); 146 147 test('the local spool is released once the log is archived', opts, async () => { 148 await withS3Conductor({}, async (h) => { 149 const job = await jobWithOutput(h, { log: 'output\n', artifact: null }); 150 151 const spool = h.services.logs.spoolPath(job.run_id, job.id); 152 assert.ok(await fs.stat(spool).catch(() => null), 'the spool should exist while running'); 153 154 await h.app.inject({ 155 method: 'POST', 156 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`, 157 headers: { ...h.auth, 'content-type': 'application/json' }, 158 payload: JSON.stringify({ success: true, exit_code: 0 }), 159 }); 160 161 // Storage is the durable copy; keeping the spool too would grow 162 // without bound. 163 assert.equal(await fs.stat(spool).catch(() => null), null, 'the spool should be gone'); 164 }); 165 }); 166 167 test('an archived log still serves incremental ranges', opts, async () => { 168 await withS3Conductor({}, async (h) => { 169 const job = await jobWithOutput(h, { log: 'first line\nsecond line\n', artifact: null }); 170 171 await h.app.inject({ 172 method: 'POST', 173 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`, 174 headers: { ...h.auth, 'content-type': 'application/json' }, 175 payload: JSON.stringify({ success: true, exit_code: 0 }), 176 }); 177 178 const tail = await h.app.inject({ 179 method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log?offset=11`, 180 }); 181 assert.equal(tail.body, 'second line\n'); 182 }); 183 }); 184 185 test('the interface renders a log that lives in the bucket', opts, async () => { 186 await withS3Conductor({}, async (h) => { 187 const job = await jobWithOutput(h, { log: 'rendered from s3\n', artifact: null }); 188 189 await h.app.inject({ 190 method: 'POST', 191 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/done`, 192 headers: { ...h.auth, 'content-type': 'application/json' }, 193 payload: JSON.stringify({ success: true, exit_code: 0 }), 194 }); 195 196 const page = await h.app.inject({ method: 'GET', url: `/jobs/${encodeURIComponent(job.id)}` }); 197 assert.equal(page.statusCode, 200); 198 assert.match(page.body, /rendered from s3/); 199 }); 200 }); 201 202 test('a private run in a bucket is still not readable by a stranger', opts, async () => { 203 await withS3Conductor({ visibility: 'private' }, async (h) => { 204 const job = await jobWithOutput(h); 205 const detail = await h.app.inject({ 206 method: 'GET', 207 url: `/api/jobs/${encodeURIComponent(job.id)}`, 208 headers: await h.login().catch(() => ({})), 209 }); 210 211 // Anonymous access is refused before any URL is ever signed, so the 212 // object store is never asked and no presigned link is handed out. 213 const anonymous = await h.app.inject({ method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}` }); 214 assert.equal(anonymous.statusCode, 404); 215 216 const anonymousLog = await h.app.inject({ 217 method: 'GET', url: `/api/jobs/${encodeURIComponent(job.id)}/log`, 218 }); 219 assert.equal(anonymousLog.statusCode, 404); 220 assert.equal(detail.statusCode, 404); 221 }); 222 }); 223 224 test('a large artifact streams through to the bucket', opts, async () => { 225 await withS3Conductor({}, async (h, { store }) => { 226 await h.trigger({ sha: h.sha, ref: 'refs/heads/main' }); 227 const job = (await h.poll({ arches: 'x86_64' })).json().job; 228 229 // Comfortably past the server's body limit for parsed payloads, which 230 // an artifact must not be subject to: build output is routinely far 231 // larger than a JSON request. 232 const payload = Buffer.alloc(3 * 1024 * 1024, 0x41); 233 234 const upload = await h.app.inject({ 235 method: 'POST', 236 url: `/api/workers/jobs/${encodeURIComponent(job.id)}/artifact`, 237 headers: { ...octet(h), 'x-artifact-path': 'dist/big.bin' }, 238 payload, 239 }); 240 assert.equal(upload.statusCode, 200, upload.body); 241 assert.equal(upload.json().size, payload.length); 242 243 const stored = await store.head(keys.artifact(job.run_id, job.id, 'dist/big.bin')); 244 assert.equal(stored.size, payload.length); 245 }); 246 });