Add support for analysing fastly log files (eclipse-openvsx#1599)

* add support for analysing fastly log files

* rename operation to method

* refactor log services and support configurable cron schedule

* add logging

* fix fastly log parser: search for start of json content

* update test data

Author: netomi

…storage/log/AwsDownloadCountService.java …storage/log/AwsDownloadCountHandler.javaserver/src/main/java/org/eclipse/openvsx/storage/log/AwsDownloadCountService.java renamed to server/src/main/java/org/eclipse/openvsx/storage/log/AwsDownloadCountHandler.java server/src/main/java/org/eclipse/openvsx/storage/log/AwsDownloadCountService.java renamed to server/src/main/java/org/eclipse/openvsx/storage/log/AwsDownloadCountHandler.java

@@ -12,21 +12,22 @@
 import org.apache.commons.lang3.StringUtils;
 import org.eclipse.openvsx.entities.Extension;
 import org.eclipse.openvsx.entities.FileResource;
+import org.eclipse.openvsx.migration.HandlerJobRequest;
 import org.eclipse.openvsx.storage.AwsStorageService;
 import org.eclipse.openvsx.util.TempFile;
 import org.jobrunr.jobs.annotations.Job;
-import org.jobrunr.jobs.annotations.Recurring;
+import org.jobrunr.jobs.lambdas.JobRequestHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StopWatch;
 import org.springframework.web.util.UriUtils;
-import software.amazon.awssdk.auth.credentials.*;
 import software.amazon.awssdk.core.sync.ResponseTransformer;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.*;
 
+import javax.annotation.PostConstruct;
 import java.io.*;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -41,17 +42,22 @@
 /**
  * Pulls logs from an Amazon S3 bucket, extracts downloads from the logs and updates download counts in the database.
  * <p>
- * Currently only log files uploaded by Amazon CloudFront are supported.
+ * The following log file formats are supported:
+ * <ul>
+ *     <li>cloudfront</li>
+ *     <li>fastly</li>
+ * </ul>
  * <p>
- * Links:
+ * See
  * <ul>
- *     <li><a href="https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/standard-logging.html">CloudFront standard logging</a></li>
- *     <li><a href="https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/standard-logs-reference.html">CloudFront log format</a></li>
+ *   <li><a href="https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/standard-logging.html">CloudFront standard logging</a></li>
+ *   <li><a href="https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/standard-logs-reference.html">CloudFront log format</a></li>
+ *   <li><a href="https://www.fastly.com/documentation/guides/integrations/streaming-logs/custom-log-formats/">Fastly custom log format</a></li>
  * </ul>
  */
 @Component
-public class AwsDownloadCountService {
-    private final Logger logger = LoggerFactory.getLogger(AwsDownloadCountService.class);
+public class AwsDownloadCountHandler implements JobRequestHandler<HandlerJobRequest<?>> {
+    private final Logger logger = LoggerFactory.getLogger(AwsDownloadCountHandler.class);
 
     private static final String LOG_LOCATION_PREFIX = "AWSLogs/";
     private static final int MAX_KEYS = 100;
@@ -65,11 +71,36 @@ public class AwsDownloadCountService {
     @Value("${ovsx.logs.aws.log-location-prefix:" + LOG_LOCATION_PREFIX + "}")
     String logLocationPrefix;
 
-    public AwsDownloadCountService(AwsStorageService awsStorageService, DownloadCountProcessor processor) {
+    @Value("${ovsx.logs.aws.format:cloudfront}")
+    String logFormat;
+
+    @Value("${ovsx.logs.aws.cron:0 10 * * * *}")
+    String cronSchedule;
+
+    LogFileParser logFileParser;
+
+    public AwsDownloadCountHandler(AwsStorageService awsStorageService, DownloadCountProcessor processor) {
         this.awsStorageService = awsStorageService;
         this.processor = processor;
     }
 
+    @PostConstruct
+    public void initialize() {
+        logFileParser = switch (logFormat.toLowerCase()) {
+            case "cloudfront" -> new CloudFrontLogFileParser();
+            case "fastly" -> new FastlyLogFileParser();
+            default -> throw new IllegalArgumentException("unsupported log file format '" + logFormat + "'");
+        };
+    }
+
+    public String getRecurringJobId() {
+        return "update-aws-download-counts";
+    }
+
+    public String getCronSchedule() {
+        return cronSchedule;
+    }
+
     /**
      * Indicates whether the download service is enabled by application config.
      */
@@ -82,11 +113,11 @@ private S3Client getS3Client() {
     }
 
     /**
-     * Task scheduled once per hour to pull logs from AWS S3 Storage and update extension download counts.
+     * Scheduled task to pull logs from AWS S3 Storage and update extension download counts.
      */
+    @Override
     @Job(name = "Update AWS Download Counts", retries = 0)
-    @Recurring(id = "update-aws-download-counts", cron = "0 10 * * * *", zoneId = "UTC")
-    public void updateDownloadCounts() {
+    public void run(HandlerJobRequest<?> jobRequest) throws Exception {
         if (!isEnabled()) {
             return;
         }
@@ -193,16 +224,14 @@ private Map<String, Integer> processLogFile(String fileName) throws IOException
             var lines = reader.lines().iterator();
             while (lines.hasNext()) {
                 var line = lines.next();
-                if (line.startsWith("#")) {
+
+                var record = logFileParser.parse(line);
+                if (record == null) {
                     continue;
                 }
 
-                // Format:
-                // date	time x-edge-location sc-bytes c-ip cs-method cs(Host) cs-uri-stem sc-status	cs(Referer)	cs(User-Agent) cs-uri-query cs(Cookie) x-edge-result-type	x-edge-request-id	x-host-header	cs-protocol	cs-bytes	time-taken	x-forwarded-for	ssl-protocol	ssl-cipher	x-edge-response-result-type	cs-protocol-version	fle-status	fle-encrypted-fields	c-port	time-to-first-byte	x-edge-detailed-result-type	sc-content-type	sc-content-len	sc-range-start	sc-range-end
-                var components = line.split("[ \t]+");
-
-                if (isGetOperation(components) && isStatusOk(components) && isExtensionPackageUri(components)) {
-                    var uri = components[7];
+                if (isGetOperation(record) && isStatusOk(record) && isExtensionPackageUri(record)) {
+                    var uri = record.url();
                     var uriComponents = uri.split("/");
                     var vsixFile = UriUtils.decode(uriComponents[uriComponents.length - 1], StandardCharsets.UTF_8).toUpperCase();
                     fileCounts.merge(vsixFile, 1, Integer::sum);
@@ -212,16 +241,16 @@ private Map<String, Integer> processLogFile(String fileName) throws IOException
         }
     }
 
-    private boolean isGetOperation(String[] components) {
-        return components[5].equalsIgnoreCase("GET");
+    private boolean isGetOperation(LogRecord record) {
+        return record.method().equalsIgnoreCase("GET");
     }
 
-    private boolean isStatusOk(String[] components) {
-        return Integer.parseInt(components[8]) == 200;
+    private boolean isStatusOk(LogRecord record) {
+        return record.status() == 200;
     }
 
-    private boolean isExtensionPackageUri(String[] components) {
-        return components[7].endsWith(".vsix");
+    private boolean isExtensionPackageUri(LogRecord record) {
+        return record.url().endsWith(".vsix");
     }
 
     private TempFile downloadFile(String objectKey) throws IOException {

…orage/log/AzureDownloadCountService.java …orage/log/AzureDownloadCountHandler.javaserver/src/main/java/org/eclipse/openvsx/storage/log/AzureDownloadCountService.java renamed to server/src/main/java/org/eclipse/openvsx/storage/log/AzureDownloadCountHandler.java server/src/main/java/org/eclipse/openvsx/storage/log/AzureDownloadCountService.java renamed to server/src/main/java/org/eclipse/openvsx/storage/log/AzureDownloadCountHandler.java

@@ -21,9 +21,11 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.StringUtils;
 import org.eclipse.openvsx.entities.FileResource;
+import org.eclipse.openvsx.migration.HandlerJobRequest;
 import org.eclipse.openvsx.util.TempFile;
 import org.jobrunr.jobs.annotations.Job;
 import org.jobrunr.jobs.annotations.Recurring;
+import org.jobrunr.jobs.lambdas.JobRequestHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -51,9 +53,9 @@
  * and updates download counts in the database.
  */
 @Component
-public class AzureDownloadCountService {
+public class AzureDownloadCountHandler implements JobRequestHandler<HandlerJobRequest<?>> {
 
-    protected final Logger logger = LoggerFactory.getLogger(AzureDownloadCountService.class);
+    protected final Logger logger = LoggerFactory.getLogger(AzureDownloadCountHandler.class);
 
     private final DownloadCountProcessor processor;
     private BlobContainerClient containerClient;
@@ -75,10 +77,21 @@ public class AzureDownloadCountService {
     @Value("${ovsx.storage.azure.blob-container:openvsx-resources}")
     String storageBlobContainer;
 
-    public AzureDownloadCountService(DownloadCountProcessor processor) {
+    @Value("${ovsx.logs.azure.cron:0 5 * * * *}")
+    String cronSchedule;
+
+    public AzureDownloadCountHandler(DownloadCountProcessor processor) {
         this.processor = processor;
     }
 
+    public String getRecurringJobId() {
+        return "update-azure-download-counts";
+    }
+
+    public String getCronSchedule() {
+        return cronSchedule;
+    }
+
     /**
      * Indicates whether the download service is enabled by application config.
      */
@@ -96,8 +109,7 @@ public boolean isEnabled() {
      * Task scheduled once per hour to pull logs from Azure Blob Storage and update extension download counts.
      */
     @Job(name = "Update Azure Download Counts", retries = 0)
-    @Recurring(id = "update-azure-download-counts", cron = "0 5 * * * *", zoneId = "UTC")
-    public void updateDownloadCounts() {
+    public void run(HandlerJobRequest<?> jobRequest) throws Exception {
         if (!isEnabled()) {
             return;
         }

server/src/main/java/org/eclipse/openvsx/storage/log/CloudFrontLogFileParser.java

@@ -0,0 +1,27 @@
+/******************************************************************************
+ * Copyright (c) 2026 Contributors to the Eclipse Foundation.
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * https://www.eclipse.org/legal/epl-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *****************************************************************************/
+package org.eclipse.openvsx.storage.log;
+
+class CloudFrontLogFileParser implements LogFileParser {
+    @Override
+    public LogRecord parse(String line) {
+        if (line.startsWith("#")) {
+            return null;
+        }
+
+        // Format:
+        // date	time x-edge-location sc-bytes c-ip cs-method cs(Host) cs-uri-stem sc-status	cs(Referer)	cs(User-Agent) cs-uri-query cs(Cookie) x-edge-result-type	x-edge-request-id	x-host-header	cs-protocol	cs-bytes	time-taken	x-forwarded-for	ssl-protocol	ssl-cipher	x-edge-response-result-type	cs-protocol-version	fle-status	fle-encrypted-fields	c-port	time-to-first-byte	x-edge-detailed-result-type	sc-content-type	sc-content-len	sc-range-start	sc-range-end
+        var components = line.split("[ \t]+");
+        return new LogRecord(components[5], Integer.parseInt(components[8]), components[7]);
+    }
+}

server/src/main/java/org/eclipse/openvsx/storage/log/DownloadCountService.java

@@ -10,24 +10,37 @@
 package org.eclipse.openvsx.storage.log;
 
 import org.eclipse.openvsx.entities.FileResource;
-import org.springframework.stereotype.Component;
+import org.eclipse.openvsx.migration.HandlerJobRequest;
+import org.jobrunr.scheduling.JobRequestScheduler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.context.event.ApplicationStartedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+
+import java.time.ZoneId;
 
 /**
  * A utility service to determine whether an optimized download count service
  * is available for a specific {@link FileResource}.
  */
-@Component
+@Service
 public class DownloadCountService {
 
-    private final AwsDownloadCountService awsDownloadCountService;
-    private final AzureDownloadCountService azureDownloadCountService;
+    protected final Logger logger = LoggerFactory.getLogger(DownloadCountService.class);
+
+    private final JobRequestScheduler scheduler;
+    private final AwsDownloadCountHandler awsDownloadCountHandler;
+    private final AzureDownloadCountHandler azureDownloadCountHandler;
 
     public DownloadCountService(
-            AwsDownloadCountService awsDownloadCountService,
-            AzureDownloadCountService azureDownloadCountService
+            JobRequestScheduler scheduler,
+            AwsDownloadCountHandler awsDownloadCountHandler,
+            AzureDownloadCountHandler azureDownloadCountHandler
     ) {
-        this.awsDownloadCountService = awsDownloadCountService;
-        this.azureDownloadCountService = azureDownloadCountService;
+        this.scheduler = scheduler;
+        this.awsDownloadCountHandler = awsDownloadCountHandler;
+        this.azureDownloadCountHandler = azureDownloadCountHandler;
     }
 
     /**
@@ -38,9 +51,36 @@ public DownloadCountService(
      */
     public boolean isEnabled(FileResource resource) {
         return switch (resource.getStorageType()) {
-            case FileResource.STORAGE_AWS -> awsDownloadCountService.isEnabled();
-            case FileResource.STORAGE_AZURE -> azureDownloadCountService.isEnabled();
+            case FileResource.STORAGE_AWS -> awsDownloadCountHandler.isEnabled();
+            case FileResource.STORAGE_AZURE -> azureDownloadCountHandler.isEnabled();
             default -> false;
         };
     }
+
+    @EventListener
+    public void applicationStarted(ApplicationStartedEvent event) {
+        if (awsDownloadCountHandler.isEnabled()) {
+            logger.info("Scheduling AWS download count handler with cron '{}'", awsDownloadCountHandler.getCronSchedule());
+            scheduler.scheduleRecurrently(
+                    awsDownloadCountHandler.getRecurringJobId(),
+                    awsDownloadCountHandler.getCronSchedule(),
+                    ZoneId.of("UTC"),
+                    new HandlerJobRequest<>(AwsDownloadCountHandler.class)
+            );
+        } else {
+            scheduler.deleteRecurringJob(awsDownloadCountHandler.getRecurringJobId());
+        }
+
+        if (azureDownloadCountHandler.isEnabled()) {
+            logger.info("Scheduling Azure download count handler with cron '{}'", azureDownloadCountHandler.getCronSchedule());
+            scheduler.scheduleRecurrently(
+                    azureDownloadCountHandler.getRecurringJobId(),
+                    azureDownloadCountHandler.getCronSchedule(),
+                    ZoneId.of("UTC"),
+                    new HandlerJobRequest<>(AwsDownloadCountHandler.class)
+            );
+        } else {
+            scheduler.deleteRecurringJob(azureDownloadCountHandler.getRecurringJobId());
+        }
+    }
 }

server/src/main/java/org/eclipse/openvsx/storage/log/FastlyLogFileParser.java

@@ -0,0 +1,76 @@
+/******************************************************************************
+ * Copyright (c) 2026 Contributors to the Eclipse Foundation.
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * https://www.eclipse.org/legal/epl-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *****************************************************************************/
+package org.eclipse.openvsx.storage.log;
+
+import com.fasterxml.jackson.core.JacksonException;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+
+class FastlyLogFileParser implements LogFileParser {
+    private final Logger logger = LoggerFactory.getLogger(FastlyLogFileParser.class);
+
+    private final ObjectMapper mapper;
+
+    public FastlyLogFileParser() {
+        this.mapper = new ObjectMapper();
+
+        SimpleModule module = new SimpleModule();
+        module.addDeserializer(LogRecord.class, new LogRecordDeserializer());
+        mapper.registerModule(module);
+    }
+
+    @Override
+    public @Nullable LogRecord parse(String line) {
+        try {
+            var jsonStartIndex = line.indexOf("{");
+            if (jsonStartIndex != -1) {
+                return mapper.readValue(line.substring(jsonStartIndex), LogRecord.class);
+            } else {
+                return null;
+            }
+        } catch (JacksonException ex) {
+            logger.error("could not parse log line '{}'", line, ex);
+            return null;
+        }
+    }
+}
+
+class LogRecordDeserializer extends StdDeserializer<LogRecord> {
+
+    public LogRecordDeserializer() {
+        this(null);
+    }
+
+    public LogRecordDeserializer(Class<?> vc) {
+        super(vc);
+    }
+
+    @Override
+    public LogRecord deserialize(JsonParser jp, DeserializationContext ctxt)
+            throws IOException {
+        JsonNode node = jp.getCodec().readTree(jp);
+        String operation = node.get("request_method").asText();
+        int status = (Integer) node.get("response_status").numberValue();
+        String url = node.get("url").asText();
+        return new LogRecord(operation, status, url);
+    }
+}