diff --git a/src/main/groovy/io/seqera/wave/proxy/ProxyClient.groovy b/src/main/groovy/io/seqera/wave/proxy/ProxyClient.groovy index c03f04254..177147bf0 100644 --- a/src/main/groovy/io/seqera/wave/proxy/ProxyClient.groovy +++ b/src/main/groovy/io/seqera/wave/proxy/ProxyClient.groovy @@ -386,6 +386,16 @@ class ProxyClient { final result = new ArrayList(20) result.add('curl') result.add('-s') + // fail with a non-zero exit status on HTTP errors, instead of writing + // the error body (or nothing at all) to stdout and exiting zero + result.add('-f') + // retry transient failures; note '--retry-all-errors' is deliberately not added because + // it would retry a response whose body was already partially piped to the upload command, + // duplicating those bytes in the uploaded object + result.add('--retry'); result.add(String.valueOf(httpConfig.retryAttempts)) + // bound the connection setup only; an upstream that connects and then stalls mid-body is + // still capped by the transfer job timeout, not by this option + result.add('--connect-timeout'); result.add(String.valueOf(httpConfig.connectTimeout.toSeconds())) result.add('-X'); result.add('GET') // copy headers for( Map.Entry entry : headers ) { diff --git a/src/main/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImpl.groovy b/src/main/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImpl.groovy index 7935d4a36..44f11416b 100644 --- a/src/main/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImpl.groovy +++ b/src/main/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImpl.groovy @@ -42,6 +42,7 @@ import jakarta.inject.Inject import jakarta.inject.Named import jakarta.inject.Singleton import software.amazon.awssdk.services.s3.S3Client +import software.amazon.awssdk.services.s3.model.DeleteObjectRequest import software.amazon.awssdk.services.s3.model.HeadObjectRequest import software.amazon.awssdk.services.s3.model.S3Exception /** @@ -113,6 +114,16 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler { } protected boolean blobExists(String blobLocation) { + return blobSize(blobLocation) != null + } + + /** + * Retrieve the size of the object stored in the object storage cache + * + * @param blobLocation The object storage path e.g. {@code s3://bucket-name/some/path} + * @return The size in bytes of the stored object or {@code null} if the object does not exist + */ + protected Long blobSize(String blobLocation) { try { final object = BucketTokenizer.from(blobLocation) final request = HeadObjectRequest @@ -121,14 +132,42 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler { .key(object.key) .build() as HeadObjectRequest // Execute the request - s3Client.headObject(request) - return true + return s3Client.headObject(request).contentLength() } catch (S3Exception e) { if (e.statusCode() != 404) { log.error "Unexpected response=${e.statusCode()} checking existence for object=${blobLocation} - cause: ${e.message}" } - return false + return null + } + catch (Exception e) { + log.error "Unexpected error checking existence for object=${blobLocation} - cause: ${e.message}", e + return null + } + } + + /** + * Delete the object stored in the object storage cache. This is used to remove an invalid + * object left behind by a failed transfer, so that it is not reported as a valid cache entry + * by {@link #blobExists(java.lang.String)} on a subsequent request. + * + * @param blobLocation The object storage path e.g. {@code s3://bucket-name/some/path} + */ + protected void deleteBlob(String blobLocation) { + try { + final object = BucketTokenizer.from(blobLocation) + final request = DeleteObjectRequest + .builder() + .bucket(object.bucket) + .key(object.key) + .build() as DeleteObjectRequest + s3Client.deleteObject(request) + log.debug "Deleted invalid blob cache object=${blobLocation}" + } + catch (Exception e) { + // the entry is marked as errored regardless, therefore a failure to clean up the + // object must not propagate out of the job completion handling + log.error "Unable to delete invalid blob cache object=${blobLocation} - cause: ${e.message}", e } } @@ -172,10 +211,16 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler { final curl = proxyService.curl(route, info.headers) final s5cmd = s5cmd(route, info) + // 'set -o pipefail' makes the pipeline fail when the curl command fails; without it + // the exit status is the one of the last command i.e. s5cmd, which happily exits zero + // after uploading an empty stream when curl failed to download the blob. + // Note 'bash' is required instead of 'sh': in the s5cmd image /bin/sh is 'dash', which + // does not implement 'pipefail' and aborts the whole script with "Illegal option -o + // pipefail" before the pipeline is even executed final command = List.of( - 'sh', + 'bash', '-c', - Escape.cli(curl) + ' | ' + Escape.cli(s5cmd) ) + 'set -o pipefail; ' + Escape.cli(curl) + ' | ' + Escape.cli(s5cmd) ) log.trace "== Blob cache transfer command: ${command.join(' ')}" return command @@ -274,12 +319,50 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler { blobStore.getBlob(job.entryKey) } + /** + * Verify the object uploaded in the object storage cache matches the expected size, to prevent + * an empty or truncated transfer from being cached and served as if it were valid. Note the + * transfer job and the Wave proxy issue two separate requests to the upstream registry, therefore + * a job can report success even when no byte was actually downloaded. + * + * @param entry The {@link BlobEntry} associated with the completed transfer + * @return An error message describing the mismatch or {@code null} when the transfer is valid + */ + protected String checkTransferredSize(BlobEntry entry) { + final uploaded = blobSize(entry.objectUri) + if( uploaded == null ) + return "Blob cache object '${entry.objectUri}' was not uploaded to the object storage".toString() + // the upstream content length is not always provided, in that case only an + // empty object can be detected as invalid + final expected = entry.contentLength + if( expected == null ) { + return uploaded == 0 + ? "Blob cache object '${entry.objectUri}' is empty".toString() + : null + } + if( uploaded != expected ) + return "Blob cache object '${entry.objectUri}' size does not match the expected content length - uploaded: ${uploaded}; expected: ${expected}".toString() + return null + } + @Override void onJobCompletion(JobSpec job, BlobEntry entry, JobState state) { + // the transfer job exit status only tells the command pipeline exited zero, it does not + // guarantee the blob bytes made it to the object storage - validate before completing + final error = state.succeeded() + ? checkTransferredSize(entry) + : state.stdout + if( error ) { + log.warn "== Blob cache transfer invalid for object '${entry.objectUri}'; operation=${job.operationName} - cause: ${error}" + // remove whatever the failed transfer left behind: a failing pipeline can still have + // uploaded an empty or truncated object, and blobExists() would then report it as a + // valid cache entry on the next request, making the bad transfer sticky + deleteBlob(entry.objectUri) + } // update the entry status - final result = state.succeeded() + final result = !error ? entry.completed(state.exitCode, state.stdout) - : entry.errored(state.stdout) + : entry.errored(error) blobStore.storeBlob(entry.getKey(), result) log.debug "== Blob cache completed for object '${entry.objectUri}'; operation=${job.operationName}; status=${result.exitStatus}; duration=${result.duration()}" } diff --git a/src/test/groovy/io/seqera/wave/proxy/ProxyClientTest.groovy b/src/test/groovy/io/seqera/wave/proxy/ProxyClientTest.groovy index f952cf4e8..66e955652 100644 --- a/src/test/groovy/io/seqera/wave/proxy/ProxyClientTest.groovy +++ b/src/test/groovy/io/seqera/wave/proxy/ProxyClientTest.groovy @@ -289,12 +289,17 @@ class ProxyClientTest extends Specification { then: cli[0] == 'curl' cli[1] == '-s' - cli[2] == '-X' - cli[3] == 'GET' - cli[4] == '-H' - cli[5] == 'content-type: application/octet-stream' - cli[6] == '-H' - cli[7] =~ /Authorization: Bearer.*/ - cli[8] == 'https://registry-1.docker.io/v2/library/hello-world/manifests/sha256:aa0cc8055b82dc2509bed2e19b275c8f463506616377219d9642221ab53cf9fe' + cli[2] == '-f' + cli[3] == '--retry' + cli[4] == String.valueOf(httpConfig.retryAttempts) + cli[5] == '--connect-timeout' + cli[6] == String.valueOf(httpConfig.connectTimeout.toSeconds()) + cli[7] == '-X' + cli[8] == 'GET' + cli[9] == '-H' + cli[10] == 'content-type: application/octet-stream' + cli[11] == '-H' + cli[12] =~ /Authorization: Bearer.*/ + cli[13] == 'https://registry-1.docker.io/v2/library/hello-world/manifests/sha256:aa0cc8055b82dc2509bed2e19b275c8f463506616377219d9642221ab53cf9fe' } } diff --git a/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest.groovy b/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest.groovy index bc8e555f5..d49cbff74 100644 --- a/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest.groovy +++ b/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest.groovy @@ -19,11 +19,17 @@ package io.seqera.wave.service.blob.impl import spock.lang.Specification +import java.time.Duration +import java.time.Instant + import io.seqera.wave.configuration.BlobCacheConfig import io.seqera.wave.core.RegistryProxyService import io.seqera.wave.core.RoutePath import io.seqera.wave.model.ContainerCoordinates import io.seqera.wave.service.blob.BlobEntry +import io.seqera.wave.service.blob.BlobStateStore +import io.seqera.wave.service.job.JobSpec +import io.seqera.wave.service.job.JobState import io.seqera.wave.test.AwsS3TestContainer /** @@ -77,10 +83,97 @@ class BlobCacheServiceImplTest extends Specification implements AwsS3TestContain proxyService.curl(route, [foo:'one']) >> ['curl', '-X', 'GET', 'http://foo'] and: result == [ - 'sh', + 'bash', '-c', - "curl -X GET 'http://foo' | s5cmd --json pipe --content-type something 's3://store/blobs/docker.io/v2/library/ubuntu/manifests/sha256:aabbcc'" + "set -o pipefail; curl -X GET 'http://foo' | s5cmd --json pipe --content-type something 's3://store/blobs/docker.io/v2/library/ubuntu/manifests/sha256:aabbcc'" ] } + private static BlobEntry blobEntry(String objectUri, Long contentLength) { + final response = contentLength!=null + ? ['Content-Length': [String.valueOf(contentLength)]] + : [:] as Map> + return BlobEntry.create('http://foo', objectUri, [:], response) + } + + def 'should validate the transferred blob size' () { + given: + def OBJECT = 's3://store/blobs/foo' + def service = Spy(BlobCacheServiceImpl) + def entry = blobEntry(OBJECT, LENGTH) + + when: + def result = service.checkTransferredSize(entry) + then: + 1 * service.blobSize(OBJECT) >> UPLOADED + and: + (result != null) == ERROR + + where: + LENGTH | UPLOADED | ERROR + 100L | 100L | false + 100L | 0L | true + 100L | 50L | true + 100L | null | true + null | 100L | false + null | 0L | true + null | null | true + } + + def 'should error the blob entry when the uploaded object is empty' () { + given: + def OBJECT = 's3://store/blobs/foo' + def blobStore = Mock(BlobStateStore) + def service = Spy(new BlobCacheServiceImpl(blobStore: blobStore)) + def entry = blobEntry(OBJECT, 100L) + def job = JobSpec.transfer('1', 'operation-1', Instant.now(), Duration.ofMinutes(1)) + + when: + service.onJobCompletion(job, entry, JobState.succeeded('some logs')) + then: + 1 * service.blobSize(OBJECT) >> 0L + and: + // the invalid object is removed so it is not served as a valid cache entry + 1 * service.deleteBlob(OBJECT) >> null + and: + 1 * blobStore.storeBlob(OBJECT, { BlobEntry it -> it.state==BlobEntry.State.ERRORED && !it.succeeded() }) + } + + def 'should complete the blob entry when the uploaded object matches the content length' () { + given: + def OBJECT = 's3://store/blobs/foo' + def blobStore = Mock(BlobStateStore) + def service = Spy(new BlobCacheServiceImpl(blobStore: blobStore)) + def entry = blobEntry(OBJECT, 100L) + def job = JobSpec.transfer('1', 'operation-1', Instant.now(), Duration.ofMinutes(1)) + + when: + service.onJobCompletion(job, entry, JobState.succeeded('some logs')) + then: + 1 * service.blobSize(OBJECT) >> 100L + and: + 0 * service.deleteBlob(_) >> null + and: + 1 * blobStore.storeBlob(OBJECT, { BlobEntry it -> it.state==BlobEntry.State.COMPLETED && it.succeeded() }) + } + + def 'should not validate the object when the transfer job failed' () { + given: + def OBJECT = 's3://store/blobs/foo' + def blobStore = Mock(BlobStateStore) + def service = Spy(new BlobCacheServiceImpl(blobStore: blobStore)) + def entry = blobEntry(OBJECT, 100L) + def job = JobSpec.transfer('1', 'operation-1', Instant.now(), Duration.ofMinutes(1)) + + when: + service.onJobCompletion(job, entry, JobState.failed(1, 'curl failed')) + then: + 0 * service.blobSize(_) >> null + and: + // a failed pipeline can still have uploaded a partial object, remove it + 1 * service.deleteBlob(OBJECT) >> null + and: + 1 * blobStore.storeBlob(OBJECT, { BlobEntry it -> it.state==BlobEntry.State.ERRORED && it.logs=='curl failed' }) + } + } diff --git a/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest2.groovy b/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest2.groovy index 94c8052cf..68ceb6df6 100644 --- a/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest2.groovy +++ b/src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest2.groovy @@ -132,7 +132,7 @@ class BlobCacheServiceImplTest2 extends Specification implements AwsS3TestContai def store = Mock(BlobStateStore) def blob = BlobEntry.create('http://some/blob','s3://some/blob', [:], [:]) def config = new BlobCacheConfig(statusDelay: Duration.ofSeconds(2)) - def service = new BlobCacheServiceImpl(blobStore: store, blobConfig: config) + def service = Spy(new BlobCacheServiceImpl(blobStore: store, blobConfig: config)) def job = JobSpec.transfer('job-id', 'foo', Instant.now(), Duration.ofMinutes(1)) def failed = new JobState(JobState.Status.FAILED, 1, 'Oops') def ok = new JobState(JobState.Status.SUCCEEDED, 0, 'done') @@ -140,11 +140,17 @@ class BlobCacheServiceImplTest2 extends Specification implements AwsS3TestContai when: service.onJobCompletion(job, blob, failed) then: + // a failed pipeline can still have uploaded a partial object, remove it + 1 * service.deleteBlob('s3://some/blob') >> null + and: 1 * store.storeBlob(blob.getKey(), _ as BlobEntry) >> { id, BlobEntry info -> info.state==BlobEntry.State.ERRORED } when: service.onJobCompletion(job, blob, ok) then: + // the transferred object is validated before the entry is marked as completed + 1 * service.blobSize('s3://some/blob') >> 100L + and: 1 * store.storeBlob(blob.getKey(), _ as BlobEntry) >> { id, BlobEntry info -> info.state==BlobEntry.State.COMPLETED } }