<commit_before>package com.timgroup.eventstore.archiver;

import com.amazonaws.util.IOUtils;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import com.timgroup.eventstore.api.EventRecord;
import com.timgroup.eventstore.api.EventSource;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;
import com.timgroup.eventsubscription.Deserializer;
import com.timgroup.eventsubscription.Event;
import com.timgroup.eventsubscription.EventHandler;
import com.timgroup.eventsubscription.EventSubscription;
import com.timgroup.eventsubscription.SubscriptionBuilder;
import com.timgroup.remotefilestorage.s3.S3UploadableStorageForInputStream;
import com.timgroup.tucker.info.Component;
import com.timgroup.tucker.info.Report;
import com.timgroup.tucker.info.Status;
import com.timgroup.tucker.info.component.SimpleValueComponent;

import javax.annotation.Nonnull;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.zip.GZIPOutputStream;

import static com.timgroup.tucker.info.Status.INFO;
import static java.lang.String.format;
import static java.nio.ByteOrder.LITTLE_ENDIAN;
import static java.util.stream.Collectors.toList;

public class S3Archiver {

    public static final String DEFAULT_MONITORING_PREFIX = "tg-eventstore-s3-archiver";

    private final EventSource liveEventSource;
    private final EventSubscription eventSubscription;

    private final String eventStoreId;
    private final BatchingPolicy batchingPolicy;
    private final S3ArchiveMaxPositionFetcher maxPositionFetcher;
    private final S3ArchiveKeyFormat batchS3ObjectKeyFormat;
    private final BatchingUploadHandler batchingUploadHandler;

    private final Clock clock;

    private final String applicationName;
    private final SimpleValueComponent checkpointPositionComponent;
    private final AtomicLong maxPositionInArchive = new AtomicLong();
    private final Timer s3ListingTimer;
    private final Timer s3UploadTimer;
    private final Histogram uncompressedSizeMetrics;
    private final Histogram compressedSizeMetrics;
    private final String monitoringPrefix;

    private S3Archiver(EventSource liveEventSource,
                       S3UploadableStorageForInputStream output,
                       String eventStoreId,
                       SubscriptionBuilder subscriptionBuilder,
                       BatchingPolicy batchingPolicy,
                       Optional<Long> maxPositionInArchiveOnStartup,
                       S3ArchiveMaxPositionFetcher maxPositionFetcher,
                       String applicationName,
                       MetricRegistry metricRegistry,
                       String monitoringPrefix,
                       Clock clock)
    {
        this.liveEventSource = liveEventSource;
        this.eventStoreId = eventStoreId;
        this.monitoringPrefix = monitoringPrefix;
        this.batchingPolicy = batchingPolicy;
        this.applicationName = applicationName;
        this.batchS3ObjectKeyFormat = new S3ArchiveKeyFormat(eventStoreId);
        this.maxPositionFetcher = maxPositionFetcher;
        this.clock = clock;

        this.checkpointPositionComponent =  new SimpleValueComponent(this.monitoringPrefix + "-checkpoint-position",
                "Checkpoint position that archiver resumed from on startup");
        this.checkpointPositionComponent.updateValue(INFO, maxPositionInArchiveOnStartup);

        this.batchingUploadHandler = new BatchingUploadHandler(batchingPolicy, output);
        this.eventSubscription = subscriptionBuilder
                .readingFrom(liveEventSource.readAll(), convertPosition(maxPositionInArchiveOnStartup))
                .deserializingUsing(Deserializer.applying(EventRecordHolder::new))
                .publishingTo(batchingUploadHandler)
                .withMaxInitialReplayDuration(Duration.ofMinutes(30))
                .build();

        this.maxPositionInArchive.set(maxPositionInArchiveOnStartup.orElse(0L));
        metricRegistry.gauge(this.monitoringPrefix + ".archive.max_position", () -> maxPositionInArchive::get);
        metricRegistry.gauge(this.monitoringPrefix + ".archive.events_awaiting_upload", () -> batchingUploadHandler.batch::size);
        this.s3ListingTimer = metricRegistry.timer(this.monitoringPrefix + ".archive.list");
        this.s3UploadTimer = metricRegistry.timer(this.monitoringPrefix + ".archive.upload");
        this.uncompressedSizeMetrics = metricRegistry.histogram(this.monitoringPrefix + ".archive.batch.uncompressed_size_bytes");
        this.compressedSizeMetrics   = metricRegistry.histogram(this.monitoringPrefix + ".archive.batch.compressed_size_bytes");
    }

    public static S3Archiver newS3Archiver(EventSource liveEventSource, S3UploadableStorageForInputStream output,
            String eventStoreId, SubscriptionBuilder subscriptionBuilder, BatchingPolicy batchingPolicy,
            S3ArchiveMaxPositionFetcher maxPositionFetcher, String applicationName, MetricRegistry metricRegistry,
            String monitoringPrefix, Clock clock) {
        return new S3Archiver(liveEventSource, output, eventStoreId, subscriptionBuilder, batchingPolicy, maxPositionFetcher.maxPosition(),
                maxPositionFetcher, applicationName, metricRegistry, monitoringPrefix, clock);
    }

    private static String hostname() {
        try {
            return InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            return "localhost";
        }
    }

    private Position convertPosition(Optional<Long> positionInArchive) {
        Long maxPositionInArchive = positionInArchive.orElse(0L);
        return liveEventSource.readAll().storePositionCodec().deserializePosition(String.valueOf(maxPositionInArchive));
    }

    public void start() {
        this.eventSubscription.start();
    }

    public void stop() {
        this.eventSubscription.stop();
    }

    public Collection<Component> monitoring() {
        List<Component> components = new ArrayList<>();
        components.addAll(liveEventSource.monitoring());
        components.addAll(eventSubscription.statusComponents());
        components.add(new ArchiveStalenessComponent());
        components.add(checkpointPositionComponent);
        components.addAll(batchingUploadHandler.monitoring());

        return components.stream().map(c -> c.withStatusNoWorseThan(Status.WARNING)).collect(toList());
    }

    public String getEventStoreId() {
        return eventStoreId;
    }

    private final class ArchiveStalenessComponent extends Component {

        ArchiveStalenessComponent() {
            super(monitoringPrefix + "-staleness", "Is archive up to date?");
        }

        @Override
        public Report getReport() {
            Optional<S3ArchivePosition> maxPositionInArchive = maxPositionInArchive();
            Optional<Long> archivePosition = maxPositionInArchive.map(pos -> pos.value);
            Optional<Long> livePosition = lastEventInLiveEventStore()
                    .map(ResolvedEvent::position)
                    .map(liveEventSource.readAll().storePositionCodec()::serializePosition)
                    .map(Long::valueOf);

            boolean isStale = batchingPolicy.isStale(maxPositionInArchive, lastEventInLiveEventStore(), liveEventSource.readAll().storePositionCodec());

            String value = format("%s%nmax_position in live=%s%nmax_position in archive=%s",
                    isStale ? "Archive is stale compared to live event store" : "Archive is up to date with live event store",
                    livePosition.map(Object::toString).orElse("[none]"),
                    archivePosition.map(Object::toString).orElse("[none]")
            );
            return isStale ? new Report(Status.WARNING, value) : new Report(Status.OK, value);
        }
    }

    public Optional<S3ArchivePosition> maxPositionInArchive() {
        try (Timer.Context ignored = this.s3ListingTimer.time()) {
            Optional<Long> maxPosition = this.maxPositionFetcher.maxPosition();
            this.maxPositionInArchive.set(maxPosition.orElse(0L));

            return maxPosition.map(S3ArchivePosition::new);
        }
    }

    public Optional<ResolvedEvent> lastEventInLiveEventStore() {
        return liveEventSource.readAll().readLastEvent();
    }

    public static final class EventRecordHolder implements Event {<commit_msg>gallan: emit max position in event source that's being archived as a metric<commit_after>package com.timgroup.eventstore.archiver;

import com.amazonaws.util.IOUtils;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import com.timgroup.eventstore.api.EventRecord;
import com.timgroup.eventstore.api.EventSource;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;
import com.timgroup.eventsubscription.Deserializer;
import com.timgroup.eventsubscription.Event;
import com.timgroup.eventsubscription.EventHandler;
import com.timgroup.eventsubscription.EventSubscription;
import com.timgroup.eventsubscription.SubscriptionBuilder;
import com.timgroup.remotefilestorage.s3.S3UploadableStorageForInputStream;
import com.timgroup.tucker.info.Component;
import com.timgroup.tucker.info.Report;
import com.timgroup.tucker.info.Status;
import com.timgroup.tucker.info.component.SimpleValueComponent;

import javax.annotation.Nonnull;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.zip.GZIPOutputStream;

import static com.timgroup.tucker.info.Status.INFO;
import static java.lang.String.format;
import static java.nio.ByteOrder.LITTLE_ENDIAN;
import static java.util.stream.Collectors.toList;

public class S3Archiver {

    public static final String DEFAULT_MONITORING_PREFIX = "tg-eventstore-s3-archiver";

    private final EventSource liveEventSource;
    private final EventSubscription eventSubscription;

    private final String eventStoreId;
    private final BatchingPolicy batchingPolicy;
    private final S3ArchiveMaxPositionFetcher maxPositionFetcher;
    private final S3ArchiveKeyFormat batchS3ObjectKeyFormat;
    private final BatchingUploadHandler batchingUploadHandler;

    private final Clock clock;

    private final String applicationName;
    private final SimpleValueComponent checkpointPositionComponent;
    private final AtomicLong maxPositionInArchive = new AtomicLong();
    private final AtomicLong maxPositionInEventSource = new AtomicLong();
    private final Timer s3ListingTimer;
    private final Timer s3UploadTimer;
    private final Histogram uncompressedSizeMetrics;
    private final Histogram compressedSizeMetrics;
    private final String monitoringPrefix;

    private S3Archiver(EventSource liveEventSource,
                       S3UploadableStorageForInputStream output,
                       String eventStoreId,
                       SubscriptionBuilder subscriptionBuilder,
                       BatchingPolicy batchingPolicy,
                       Optional<Long> maxPositionInArchiveOnStartup,
                       S3ArchiveMaxPositionFetcher maxPositionFetcher,
                       String applicationName,
                       MetricRegistry metricRegistry,
                       String monitoringPrefix,
                       Clock clock)
    {
        this.liveEventSource = liveEventSource;
        this.eventStoreId = eventStoreId;
        this.monitoringPrefix = monitoringPrefix;
        this.batchingPolicy = batchingPolicy;
        this.applicationName = applicationName;
        this.batchS3ObjectKeyFormat = new S3ArchiveKeyFormat(eventStoreId);
        this.maxPositionFetcher = maxPositionFetcher;
        this.clock = clock;

        this.checkpointPositionComponent =  new SimpleValueComponent(this.monitoringPrefix + "-checkpoint-position",
                "Checkpoint position that archiver resumed from on startup");
        this.checkpointPositionComponent.updateValue(INFO, maxPositionInArchiveOnStartup);

        this.batchingUploadHandler = new BatchingUploadHandler(batchingPolicy, output);
        this.eventSubscription = subscriptionBuilder
                .readingFrom(liveEventSource.readAll(), convertPosition(maxPositionInArchiveOnStartup))
                .deserializingUsing(Deserializer.applying(EventRecordHolder::new))
                .publishingTo(batchingUploadHandler)
                .withMaxInitialReplayDuration(Duration.ofMinutes(30))
                .build();

        this.maxPositionInArchive.set(maxPositionInArchiveOnStartup.orElse(0L));
        metricRegistry.gauge(this.monitoringPrefix + ".archive.max_position", () -> maxPositionInArchive::get);
        metricRegistry.gauge(this.monitoringPrefix + ".event_source.max_position", () -> maxPositionInEventSource::get);
        metricRegistry.gauge(this.monitoringPrefix + ".archive.events_awaiting_upload", () -> batchingUploadHandler.batch::size);
        this.s3ListingTimer = metricRegistry.timer(this.monitoringPrefix + ".archive.list");
        this.s3UploadTimer = metricRegistry.timer(this.monitoringPrefix + ".archive.upload");
        this.uncompressedSizeMetrics = metricRegistry.histogram(this.monitoringPrefix + ".archive.batch.uncompressed_size_bytes");
        this.compressedSizeMetrics   = metricRegistry.histogram(this.monitoringPrefix + ".archive.batch.compressed_size_bytes");
    }

    public static S3Archiver newS3Archiver(EventSource liveEventSource, S3UploadableStorageForInputStream output,
            String eventStoreId, SubscriptionBuilder subscriptionBuilder, BatchingPolicy batchingPolicy,
            S3ArchiveMaxPositionFetcher maxPositionFetcher, String applicationName, MetricRegistry metricRegistry,
            String monitoringPrefix, Clock clock) {
        return new S3Archiver(liveEventSource, output, eventStoreId, subscriptionBuilder, batchingPolicy, maxPositionFetcher.maxPosition(),
                maxPositionFetcher, applicationName, metricRegistry, monitoringPrefix, clock);
    }

    private static String hostname() {
        try {
            return InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            return "localhost";
        }
    }

    private Position convertPosition(Optional<Long> positionInArchive) {
        Long maxPositionInArchive = positionInArchive.orElse(0L);
        return liveEventSource.readAll().storePositionCodec().deserializePosition(String.valueOf(maxPositionInArchive));
    }

    public void start() {
        this.eventSubscription.start();
    }

    public void stop() {
        this.eventSubscription.stop();
    }

    public Collection<Component> monitoring() {
        List<Component> components = new ArrayList<>();
        components.addAll(liveEventSource.monitoring());
        components.addAll(eventSubscription.statusComponents());
        components.add(new ArchiveStalenessComponent());
        components.add(checkpointPositionComponent);
        components.addAll(batchingUploadHandler.monitoring());

        return components.stream().map(c -> c.withStatusNoWorseThan(Status.WARNING)).collect(toList());
    }

    public String getEventStoreId() {
        return eventStoreId;
    }

    private final class ArchiveStalenessComponent extends Component {

        ArchiveStalenessComponent() {
            super(monitoringPrefix + "-staleness", "Is archive up to date?");
        }

        @Override
        public Report getReport() {
            Optional<S3ArchivePosition> maxPositionInArchive = maxPositionInArchive();
            Optional<Long> archivePosition = maxPositionInArchive.map(pos -> pos.value);
            Optional<Long> livePosition = lastEventInLiveEventStore()
                    .map(ResolvedEvent::position)
                    .map(liveEventSource.readAll().storePositionCodec()::serializePosition)
                    .map(Long::valueOf);

            livePosition.ifPresent(maxPositionInEventSource::set);

            boolean isStale = batchingPolicy.isStale(maxPositionInArchive, lastEventInLiveEventStore(), liveEventSource.readAll().storePositionCodec());

            String value = format("%s%nmax_position in live=%s%nmax_position in archive=%s",
                    isStale ? "Archive is stale compared to live event store" : "Archive is up to date with live event store",
                    livePosition.map(Object::toString).orElse("[none]"),
                    archivePosition.map(Object::toString).orElse("[none]")
            );
            return isStale ? new Report(Status.WARNING, value) : new Report(Status.OK, value);
        }
    }

    public Optional<S3ArchivePosition> maxPositionInArchive() {
        try (Timer.Context ignored = this.s3ListingTimer.time()) {
            Optional<Long> maxPosition = this.maxPositionFetcher.maxPosition();
            this.maxPositionInArchive.set(maxPosition.orElse(0L));

            return maxPosition.map(S3ArchivePosition::new);
        }
    }

    public Optional<ResolvedEvent> lastEventInLiveEventStore() {
        return liveEventSource.readAll().readLastEvent();
    }

    public static final class EventRecordHolder implements Event {
{'fraction_non_alphanumeric': 0.08332880360928412, 'fraction_numerical': 0.003261401315431864, 'mean_word_length': 3.6156547917711994, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3141762', 'n_tokens_mistral': 4801, 'n_tokens_neox': 4642, 'n_words': 898}