-
Notifications
You must be signed in to change notification settings - Fork 51
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement batch submission for traces (#469)
- Loading branch information
1 parent
043765b
commit 4687cf3
Showing
29 changed files
with
532 additions
and
177 deletions.
There are no files selected for viewing
22 changes: 22 additions & 0 deletions
22
docker/controller-node/60-add-script-signature-approvals.groovy
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| import org.jenkinsci.plugins.scriptsecurity.scripts.ScriptApproval | ||
|
|
||
| // needed by the stress-test-traces-submit pipeline | ||
| def scriptApproval = ScriptApproval.get() | ||
| def signatures = [ | ||
| "field org.datadog.jenkins.plugins.datadog.traces.write.TraceWriter asyncWriter", | ||
| "field org.datadog.jenkins.plugins.datadog.util.AsyncWriter queue", | ||
| "method java.util.concurrent.BlockingQueue remainingCapacity", | ||
| "new java.util.concurrent.atomic.AtomicBoolean boolean", | ||
| "method java.util.concurrent.atomic.AtomicBoolean get", | ||
| "method java.util.concurrent.atomic.AtomicBoolean set boolean", | ||
| "staticMethod org.datadog.jenkins.plugins.datadog.traces.write.TraceWriterFactory getTraceWriter", | ||
| ] | ||
|
|
||
| signatures.each { signature -> | ||
| if (!scriptApproval.getPendingSignatures().any { it.signature == signature }) { | ||
| scriptApproval.approveSignature(signature) | ||
| println "Approved signature: $signature" | ||
| } else { | ||
| println "Signature already pending or approved: $signature" | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| import java.util.concurrent.atomic.AtomicBoolean | ||
|
|
||
| node { | ||
| def numBuilds = params.NUM_BUILDS ?: 10 | ||
| def numSteps = params.NUM_STEPS ?: 100 | ||
| def jobComplete = new AtomicBoolean(false) | ||
| def parallelStages = [:] | ||
| parallelStages['Generate jobs'] = { | ||
| stage('Generate jobs') { | ||
| def builds = [:] | ||
| // Loop to create parallel jobs | ||
| for (int i = 1; i <= numBuilds; i++) { | ||
| def jobIndex = i | ||
| builds["Job-${jobIndex}"] = { | ||
| echo "Starting Job ${jobIndex}" | ||
| // Inner loop to create steps within each job | ||
| for (int j = 1; j <= numSteps; j++) { | ||
| echo "Executing step ${j} in Job ${jobIndex}" | ||
| // Execute a shell command to echo random characters | ||
| sh "echo ${UUID.randomUUID()}" | ||
| } | ||
| echo "Finished Load Job ${jobIndex}" | ||
| } | ||
| } | ||
| // Execute all jobs in parallel | ||
| parallel builds | ||
| jobComplete.set(true) | ||
| } | ||
| } | ||
| parallelStages['Print traces queue capacity'] = { | ||
| stage('Print traces queue capacity') { | ||
| script { | ||
| waitUntil { | ||
| echo "Remaining traces queue capacity ${org.datadog.jenkins.plugins.datadog.traces.write.TraceWriterFactory.getTraceWriter().asyncWriter.queue.remainingCapacity()}" | ||
| sleep time: 1, unit: 'SECONDS' | ||
| return jobComplete.get() | ||
| } | ||
| } | ||
| } | ||
| } | ||
| parallel parallelStages | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
71 changes: 71 additions & 0 deletions
71
src/main/java/org/datadog/jenkins/plugins/datadog/clients/CompressedBatchSender.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package org.datadog.jenkins.plugins.datadog.clients; | ||
|
|
||
| import net.sf.json.JSONObject; | ||
| import java.io.ByteArrayOutputStream; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Collection; | ||
| import java.util.Map; | ||
| import java.util.function.Function; | ||
| import java.util.logging.Logger; | ||
| import java.util.zip.GZIPOutputStream; | ||
|
|
||
| public class CompressedBatchSender<T> implements JsonPayloadSender<T> { | ||
|
|
||
| private static final Logger logger = Logger.getLogger(CompressedBatchSender.class.getName()); | ||
|
|
||
| private static final byte[] BEGIN_JSON_ARRAY = "[".getBytes(StandardCharsets.UTF_8); | ||
| private static final byte[] END_JSON_ARRAY = "]".getBytes(StandardCharsets.UTF_8); | ||
| private static final byte[] COMMA = ",".getBytes(StandardCharsets.UTF_8); | ||
|
|
||
| private final HttpClient httpClient; | ||
| private final String url; | ||
| private final Map<String, String> headers; | ||
| private final int batchLimitBytes; | ||
| private final Function<T, JSONObject> payloadToJson; | ||
|
|
||
| public CompressedBatchSender(HttpClient httpClient, | ||
| String url, | ||
| Map<String, String> headers, | ||
| int batchLimitBytes, | ||
| Function<T, JSONObject> payloadToJson) { | ||
| this.httpClient = httpClient; | ||
| this.url = url; | ||
| this.headers = headers; | ||
| this.batchLimitBytes = batchLimitBytes; | ||
| this.payloadToJson = payloadToJson; | ||
| } | ||
|
|
||
| @Override | ||
| public void send(Collection<T> payloads) throws Exception { | ||
| ByteArrayOutputStream request = new ByteArrayOutputStream(); | ||
| GZIPOutputStream gzip = new GZIPOutputStream(request); | ||
| // the backend checks the size limit against the uncompressed body of the request | ||
| int uncompressedRequestLength = 0; | ||
|
|
||
| for (T payload : payloads) { | ||
| JSONObject json = payloadToJson.apply(payload); | ||
| byte[] body = json.toString().getBytes(StandardCharsets.UTF_8); | ||
| if (body.length + 2 > batchLimitBytes) { // + 2 is for array beginning and end: [<payload>] | ||
| logger.severe("Dropping a payload because size (" + body.length + ") exceeds the allowed limit of " + batchLimitBytes); | ||
| continue; | ||
| } | ||
|
|
||
| if (uncompressedRequestLength + body.length + 2 > batchLimitBytes) { // + 2 is for comma and array end: ,<payload>] | ||
| gzip.write(END_JSON_ARRAY); | ||
| gzip.close(); | ||
| httpClient.post(url, headers, "application/json", request.toByteArray(), Function.identity()); | ||
| request = new ByteArrayOutputStream(); | ||
| gzip = new GZIPOutputStream(request); | ||
| uncompressedRequestLength = 0; | ||
| } | ||
|
|
||
| gzip.write(uncompressedRequestLength == 0 ? BEGIN_JSON_ARRAY : COMMA); | ||
| gzip.write(body); | ||
| uncompressedRequestLength += body.length + 1; | ||
| } | ||
|
|
||
| gzip.write(END_JSON_ARRAY); | ||
| gzip.close(); | ||
| httpClient.post(url, headers, "application/json", request.toByteArray(), Function.identity()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.