forked from opensearch-project/opensearch-migrations
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into ReplayerCodeSmells
Signed-off-by: Greg Schohn <[email protected]> # Conflicts: # TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/datahandlers/PayloadAccessFaultingMap.java
- Loading branch information
Showing
37 changed files
with
1,131 additions
and
543 deletions.
There are no files selected for viewing
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
24 changes: 24 additions & 0 deletions
24
TrafficCapture/dockerSolution/src/main/docker/otelCollector/README.md
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,24 @@ | ||
| ## Monitoring Progress via Instrumentation | ||
|
|
||
| The replayer and capture proxy (if started with the `--otelCollectorEndpoint` argument) emit metrics through an | ||
| otel-collector endpoint, which is deployed within Migrations Assistant tasks as a sidecar container. The | ||
| otel-collectors will publish metrics and traces to Amazon CloudWatch and AWS X-Ray. | ||
|
|
||
| Some of these metrics will show simple progress, such as bytes or records transmitted. Other records can show higher | ||
| level information, such the number of responses with status codes that match vs those that don't. To observe those, | ||
| search for `statusCodesMatch` in the CloudWatch Console. That's emitted as an attribute along with the method and | ||
| the source/target status code (rounded down to the last hundred; i.e. a status code of 201 has a 200 attribute). | ||
|
|
||
| Other metrics will show latencies, the number of requests, unique connections at a time and more. Low-level and | ||
| high-level metrics are being improved and added. For the latest information, see the | ||
| [README.md](../../../../../../coreUtilities/README.md). | ||
|
|
||
| Along with metrics, traces are emitted by the replayer and the proxy (when proxy is run with metrics enabled, e.g. by | ||
| launching with --otelCollectorEndpoint set to the otel-collector sidecar). Traces will include very granular data for | ||
| each connection, including how long the TCP connections are open, how long the source and target clusters took to send | ||
| a response, as well as other internal details that can explain the progress of each request. | ||
|
|
||
| Notice that traces for the replayer will show connections and Kafka records open, in some cases, much longer than their | ||
| representative HTTP transactions. This is because records are considered 'active' to the replayer until they are | ||
| committed and records are only committed once _all_ previous records have also been committed. Details such as that | ||
| are defensive for when further diagnosis is necessary. |
6 changes: 4 additions & 2 deletions
6
...ensearch/migrations/trafficcapture/proxyserver/testcontainers/HttpdContainerTestBase.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
7 changes: 3 additions & 4 deletions
7
...ensearch/migrations/trafficcapture/proxyserver/testcontainers/KafkaContainerTestBase.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
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
105 changes: 105 additions & 0 deletions
105
...e/trafficReplayer/src/main/java/org/opensearch/migrations/replay/AggregatedRawResult.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,105 @@ | ||
| package org.opensearch.migrations.replay; | ||
|
|
||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.util.AbstractMap; | ||
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| import org.opensearch.migrations.replay.datatypes.ByteBufList; | ||
|
|
||
| import io.netty.buffer.ByteBuf; | ||
| import io.netty.buffer.Unpooled; | ||
| import lombok.Getter; | ||
|
|
||
| public class AggregatedRawResult { | ||
| @Getter | ||
| protected final int sizeInBytes; | ||
| @Getter | ||
| protected final Duration duration; | ||
| protected final ArrayList<AbstractMap.SimpleEntry<Instant, byte[]>> packets; | ||
| @Getter | ||
| protected final Throwable error; | ||
|
|
||
| public static class Builder<B extends Builder<B>> { | ||
| protected final ArrayList<AbstractMap.SimpleEntry<Instant, byte[]>> receiptTimeAndResponsePackets; | ||
| protected final Instant startTime; | ||
| protected Throwable error; | ||
|
|
||
| public Builder(Instant startTime) { | ||
| receiptTimeAndResponsePackets = new ArrayList<>(); | ||
| this.startTime = startTime; | ||
| } | ||
|
|
||
| public AggregatedRawResult build() { | ||
| var totalBytes = getTotalBytes(); | ||
| return new AggregatedRawResult( | ||
| totalBytes, | ||
| Duration.between(startTime, Instant.now()), | ||
| receiptTimeAndResponsePackets, | ||
| error | ||
| ); | ||
| } | ||
|
|
||
| protected int getTotalBytes() { | ||
| return receiptTimeAndResponsePackets.stream().mapToInt(kvp -> kvp.getValue().length).sum(); | ||
| } | ||
|
|
||
| public B addErrorCause(Throwable t) { | ||
| error = t; | ||
| return (B) this; | ||
| } | ||
|
|
||
| public B addResponsePacket(byte[] packet) { | ||
| return (B) addResponsePacket(packet, Instant.now()); | ||
| } | ||
|
|
||
| public B addResponsePacket(byte[] packet, Instant timestamp) { | ||
| receiptTimeAndResponsePackets.add(new AbstractMap.SimpleEntry<>(timestamp, packet)); | ||
| return (B) this; | ||
| } | ||
| } | ||
|
|
||
| public AggregatedRawResult(int sizeInBytes, | ||
| Duration duration, | ||
| List<AbstractMap.SimpleEntry<Instant, byte[]>> packets, | ||
| Throwable error) | ||
| { | ||
| this.sizeInBytes = sizeInBytes; | ||
| this.duration = duration; | ||
| this.packets = packets == null ? null : new ArrayList<>(packets); | ||
| this.error = error; | ||
| } | ||
|
|
||
| public static Builder<?> builder(Instant i) { | ||
| return new Builder<>(i); | ||
| } | ||
|
|
||
| public byte[][] getCopyOfPackets() { | ||
| return packets.stream() | ||
| .map(Map.Entry::getValue) | ||
| .map(x -> Arrays.copyOf(x, x.length)) | ||
| .toArray(byte[][]::new); | ||
| } | ||
|
|
||
| public ByteBuf getResponseAsByteBuf() { | ||
| return packets == null ? Unpooled.EMPTY_BUFFER : | ||
| ByteBufList.asCompositeByteBufRetained(packets.stream() | ||
| .map(Map.Entry::getValue).map(Unpooled::wrappedBuffer)) | ||
| .asReadOnly(); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| final StringBuilder sb = new StringBuilder("IResponseSummary{"); | ||
| sb.append("responseSizeInBytes=").append(sizeInBytes); | ||
| sb.append(", responseDuration=").append(duration); | ||
| sb.append(", # of responsePackets=") | ||
| .append((this.packets == null ? "-1" : "" + this.packets.size())); | ||
| sb.append('}'); | ||
| return sb.toString(); | ||
| } | ||
|
|
||
| } |
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.