diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java index 07a21856e..d3c090612 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java @@ -47,9 +47,12 @@ * {@link Constants#FIFTYONE_COOKIE_PREFIX} or has been configured to be * shared using the constructor parameter. * 5) Not a header, cookie, session value or query string parameter. + * When all Evidence is shared, everything is included. When all query string + * parameters are shared, every query string parameter is included. * The prefix of a key is everything before its first '.', and the field * everything after it. An email address, the salt for hashed email - * identifiers, the query string and the path are never included. + * identifiers, the query string, the path and the Cookie header are never + * included, whatever the configuration. */ public class EvidenceKeyFilterShareUsage implements EvidenceKeyFilter { @@ -65,6 +68,17 @@ public class EvidenceKeyFilterShareUsage implements EvidenceKeyFilter { */ private final String sessionCookieName; + /** + * If true then every key is included, apart from those that are never + * shared. + */ + private final boolean shareAllEvidence; + + /** + * If true then every query string parameter is included. + */ + private final boolean shareAllQueryStringParameters; + /** * The content of HTTP headers in this array will not be included in the * request information sent to 51degrees. Any header names added here are @@ -113,6 +127,40 @@ public EvidenceKeyFilterShareUsage( List includedQueryStringParams, boolean includeSession, String sessionCookieName) { + this( + blockedHttpHeaders, + includedQueryStringParams, + includeSession, + sessionCookieName, + false, + false); + } + + /** + * Constructor. + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParams a list of the names of query string + * parameters that share usage should send + * to 51Degrees + * @param includeSession if true then the session cookie will be included in + * the filter + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param shareAllEvidence if true then all Evidence is included, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are included + */ + public EvidenceKeyFilterShareUsage( + List blockedHttpHeaders, + List includedQueryStringParams, + boolean includeSession, + String sessionCookieName, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { + this.shareAllEvidence = shareAllEvidence; + this.shareAllQueryStringParameters = shareAllQueryStringParameters; // The Cookie header carries every cookie, including the ones the // cookie rule leaves out. this.blockedHttpHeaders.add("cookies"); @@ -144,6 +192,10 @@ public boolean include(String key) { } } + if (shareAllEvidence) { + return true; + } + boolean result; int separator = lowerKey.indexOf(EVIDENCE_SEPERATOR); if (separator > 0) { @@ -169,8 +221,9 @@ public boolean include(String key) { case EVIDENCE_QUERY_PREFIX: // Only include query string parameters that start with the // 51Degrees prefix or have been specified in the - // constructor. - result = field.startsWith(Constants.FIFTYONE_COOKIE_PREFIX) || + // constructor, unless all of them are shared. + result = shareAllQueryStringParameters || + field.startsWith(Constants.FIFTYONE_COOKIE_PREFIX) || includedQueryStringParams.contains(field); break; default: diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java index 9240b3267..61c2c348c 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java @@ -57,6 +57,38 @@ public EvidenceKeyFilterShareUsageTracker( sessionCookieName); } + /** + * Constructor. + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParams a list of the names of query string + * parameters that share usage should send + * to 51Degrees + * @param includeSession if true then the session cookie will be included in + * the filter + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param shareAllEvidence if true then all Evidence is included, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are included + */ + public EvidenceKeyFilterShareUsageTracker( + List blockedHttpHeaders, + List includedQueryStringParams, + boolean includeSession, + String sessionCookieName, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { + super( + blockedHttpHeaders, + includedQueryStringParams, + includeSession, + sessionCookieName, + shareAllEvidence, + shareAllQueryStringParameters); + } + @Override public boolean include(String key) { if(key.equalsIgnoreCase(EVIDENCE_SESSIONID) || diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java index f53a7bf29..28506e77d 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java @@ -93,6 +93,13 @@ public class Constants { * Default lost data count for share usage before it wraps */ static final int LOST_DATA_RESET_DEFAULT = 5000000; + + /** + * While the queue is full, a message is logged for the first entry + * discarded and then once for every this many more, never one for each + * entry. + */ + static final int QUEUE_FULL_LOG_EVERY = 1000; /** * The default URL to send usage data to. */ @@ -104,4 +111,43 @@ public class Constants { */ static final String EVIDENCE_HTTPHEADER_COOKIE_SUFFIX = "cookie"; + /** + * The default number of failed sends within the failures window that + * pauses sending. + */ + public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT = 10; + + /** + * The fewest failed sends that can be set to pause sending. + */ + public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN = 1; + + /** + * The most failed sends that can be set to pause sending. + */ + public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX = 100; + + /** + * The default time in seconds over which failed sends are counted. + */ + public static final int SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT = 100; + + /** + * The default length in seconds of the first pause after failed sends. + */ + public static final double + SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT = 2.0; + + /** + * The default longest pause in seconds. + */ + public static final double + SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT = 300.0; + + /** + * The default multiplier applied to the pause each time the send tried + * when a pause ends fails. + */ + public static final double + SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT = 2.0; } diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java new file mode 100644 index 000000000..2a582d7b7 --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java @@ -0,0 +1,165 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import java.util.ArrayDeque; +import java.util.Deque; + +/** + * Decides when usage sharing may send. Sending pauses for the initial delay + * once the set number of sends fail within the window. A failed send after a + * pause multiplies the delay, up to the maximum, and a successful send ends + * the back-off. Only sending pauses, so evidence is still queued while + * paused. Each method takes the time from the caller, in milliseconds on + * the clock that {@link #now()} reads, so a change to the system time cannot + * lengthen or skip a pause. + */ +class ShareUsageBackOff { + + private final int failuresToEnterRecovery; + private final long failuresWindowMillis; + private final double initialDelaySeconds; + private final double maxDelaySeconds; + private final double multiplier; + + /** + * When each recent failed send happened, oldest first. Empty while + * backing off. + */ + private final Deque failures = new ArrayDeque<>(); + + /** + * The length of the current pause in seconds, or zero when not backing + * off. + */ + private double delaySeconds = 0; + + /** + * When sending may start again. + */ + private long resumeAt = Long.MIN_VALUE; + + /** + * @return the time in milliseconds on a monotonic clock, for the caller to + * pass to the other methods. Its zero is not a calendar time. + */ + static long now() { + return System.nanoTime() / 1000000; + } + + ShareUsageBackOff( + int failuresToEnterRecovery, + int failuresWindowSeconds, + double initialDelaySeconds, + double maxDelaySeconds, + double multiplier) { + if (failuresToEnterRecovery < + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN || + failuresToEnterRecovery > + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX) { + throw new IllegalArgumentException( + "failuresToEnterRecovery must be from " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN + + " to " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX + "."); + } + if (failuresWindowSeconds < 1) { + throw new IllegalArgumentException( + "failuresWindowSeconds must be 1 or more."); + } + if ((initialDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "initialDelaySeconds must be more than 0."); + } + if ((maxDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "maxDelaySeconds must be more than 0."); + } + if ((multiplier >= 1) == false) { + throw new IllegalArgumentException( + "multiplier must be 1 or more."); + } + this.failuresToEnterRecovery = failuresToEnterRecovery; + this.failuresWindowMillis = failuresWindowSeconds * 1000L; + this.initialDelaySeconds = initialDelaySeconds; + this.maxDelaySeconds = maxDelaySeconds; + this.multiplier = multiplier; + } + + /** + * @return true from the start of a pause until a send succeeds + */ + synchronized boolean isBackingOff() { + return delaySeconds > 0; + } + + /** + * @param now the time now in milliseconds + * @return the milliseconds left before sending may start again, or zero + * if it may start now + */ + synchronized long remaining(long now) { + return resumeAt > now ? resumeAt - now : 0; + } + + /** + * Records a failed send. + * @param now the time now in milliseconds + * @return the new pause in milliseconds if this failure started or + * lengthened one, otherwise zero + */ + synchronized long failed(long now) { + if (delaySeconds > 0) { + // The send tried when the last pause ended failed. + delaySeconds = Math.min( + delaySeconds * multiplier, + maxDelaySeconds); + } else { + failures.addLast(now); + while (failures.peekFirst() < now - failuresWindowMillis) { + failures.removeFirst(); + } + if (failures.size() < failuresToEnterRecovery) { + return 0; + } + failures.clear(); + delaySeconds = Math.min(initialDelaySeconds, maxDelaySeconds); + } + long pause = Math.max(1, Math.round(delaySeconds * 1000)); + resumeAt = now + pause; + return pause; + } + + /** + * Records a successful send. + * @return true if this ended a back-off + */ + synchronized boolean succeeded() { + if (delaySeconds == 0) { + return false; + } + delaySeconds = 0; + resumeAt = Long.MIN_VALUE; + return true; + } +} diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java index 96b240773..9e417dcfe 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java @@ -228,6 +228,23 @@ public abstract class ShareUsageBase */ protected long lostDataReset = LOST_DATA_RESET_DEFAULT; + /** + * True from the first entry discarded because the queue is full until an + * entry is next queued. While true, a message is logged only for every + * {@link Constants#QUEUE_FULL_LOG_EVERY} entries discarded. + */ + private volatile boolean queueFullReported = false; + + /** + * Decides when sending may happen after failed sends. + */ + private final ShareUsageBackOff backOff; + + /** + * Released when the element starts closing, to end any pause at once. + */ + private final CountDownLatch closing = new CountDownLatch(1); + /** * Constructor * @param logger the logger to use @@ -396,6 +413,98 @@ protected ShareUsageBase( List> ignoreDataEvidenceFilter, String sessionCookieName, Tracker tracker) { + this( + logger, + sharePercentage, + minimumEntriesPerMessage, + maximumQueueSize, + addTimeout, + takeTimeout, + repeatEvidenceIntervalMinutes, + trackSession, + shareUsageUrl, + blockedHttpHeaders, + includedQueryStringParameters, + ignoreDataEvidenceFilter, + sessionCookieName, + tracker, + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT, + Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT, + Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT, + Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT, + Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT, + false, + false); + } + + /** + * Constructor + * @param logger the logger to use + * @param sharePercentage the approximate proportion of requests to share. + * 1 = 100%, 0.5 = 50%, etc. + * @param minimumEntriesPerMessage the minimum number of request entries per + * message sent to 51Degrees + * @param maximumQueueSize the maximum number of items to hold in the queue + * at one time. This must be larger than minimum + * entries + * @param addTimeout the timeout in milliseconds to allow when attempting to + * add an item to the queue. If this timeout is exceeded + * then the item is not shared + * @param takeTimeout the timeout in milliseconds to allow when attempting + * to take an item to the queue + * @param repeatEvidenceIntervalMinutes the interval (in minutes) which is + * used to decide if repeat evidence is + * old enough to consider a new session + * @param trackSession set if the tracker should consider sessions in share + * usage + * @param shareUsageUrl the URL to send data to + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParameters a list of the names of query string + * parameters that share usage should + * send to 51Degrees + * @param ignoreDataEvidenceFilter the filter used to determine if an item + * of evidence should be ignored or not + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param tracker the {@link Tracker} to use to determine if a given + * {@link FlowData} instance should be shared or not + * @param failuresToEnterRecovery the number of failed sends within the + * failures window that pauses sending + * @param failuresWindowSeconds the time in seconds over which failed sends + * are counted + * @param initialRecoveryDelaySeconds the length in seconds of the first + * pause + * @param maxRecoveryDelaySeconds the longest a pause can be, in seconds + * @param recoveryMultiplier what the pause is multiplied by each time the + * send tried when a pause ends fails + * @param shareAllEvidence if true then all Evidence is shared, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are shared + */ + protected ShareUsageBase( + Logger logger, + double sharePercentage, + int minimumEntriesPerMessage, + int maximumQueueSize, + int addTimeout, + int takeTimeout, + int repeatEvidenceIntervalMinutes, + boolean trackSession, + String shareUsageUrl, + List blockedHttpHeaders, + List includedQueryStringParameters, + List> ignoreDataEvidenceFilter, + String sessionCookieName, + Tracker tracker, + int failuresToEnterRecovery, + int failuresWindowSeconds, + double initialRecoveryDelaySeconds, + double maxRecoveryDelaySeconds, + double recoveryMultiplier, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { super(logger, null); if (minimumEntriesPerMessage > maximumQueueSize) { @@ -410,6 +519,12 @@ protected ShareUsageBase( } evidenceCollection = new LinkedBlockingQueue<>(maximumQueueSize); + backOff = new ShareUsageBackOff( + failuresToEnterRecovery, + failuresWindowSeconds, + initialRecoveryDelaySeconds, + maxRecoveryDelaySeconds, + recoveryMultiplier); this.addTimeout = addTimeout; this.takeTimeout = takeTimeout; @@ -430,7 +545,12 @@ protected ShareUsageBase( includedQueryStringParameters.add(Constants.EVIDENCE_SEQUENCE_SUFIX); evidenceKeyFilterExclSession = new EvidenceKeyFilterShareUsage( - blockedHttpHeaders, includedQueryStringParameters, false, sessionCookieName); + blockedHttpHeaders, + includedQueryStringParameters, + false, + sessionCookieName, + shareAllEvidence, + shareAllQueryStringParameters); this.ignoreDataEvidenceFilter = ignoreDataEvidenceFilter; @@ -444,7 +564,12 @@ protected ShareUsageBase( cacheConfiguration, intervalMillis, new EvidenceKeyFilterShareUsageTracker( - blockedHttpHeaders, includedQueryStringParameters, trackSession, sessionCookieName)); + blockedHttpHeaders, + includedQueryStringParameters, + trackSession, + sessionCookieName, + shareAllEvidence, + shareAllQueryStringParameters)); } properties = new ArrayList<>(); @@ -609,14 +734,15 @@ protected void managedResourcesCleanup() { // This is set before anything is submitted, so a consumer can never // observe it as false and then exit leaving a partial batch behind. stopping = true; - // Queue a final consumer without consulting isRunning(). That check is - // subject to a time-of-check/time-of-use race: a consumer that has - // finished its loop with a sub-threshold remainder still reports as - // running until its Future flips to done, which used to stop close from - // queueing the drain that would have sent the remainder. The executor is - // single threaded, so the task queued here cannot run alongside a - // consumer that is already in flight - it runs after it - and shutdown() - // still allows an already submitted task to run to completion. + // Ends a pause at once, so closing never waits for one. Nothing is + // sent while paused. + closing.countDown(); + // Queue a final consumer without consulting isRunning(). A consumer + // that has finished its loop with a sub-threshold remainder still + // reports as running until its Future is done, so checking would skip + // the drain that sends the remainder. The executor is single threaded, + // so the task queued here runs after any consumer already in flight, + // and shutdown() still allows an already submitted task to finish. submitSendData(); try { executor.shutdown(); @@ -630,6 +756,13 @@ protected void managedResourcesCleanup() { // abandon its own wait immediately. logger.warn("Interrupted while awaiting close down"); } + if (evidenceCollection.isEmpty() == false && + backOff.remaining(ShareUsageBackOff.now()) > 0) { + logger.info( + "{} usage sharing entries were not sent, as sending was " + + "paused after failed sends when usage sharing closed.", + evidenceCollection.size()); + } } @Override @@ -670,6 +803,7 @@ private void processData(FlowData data, Map found) { getDataFromEvidence(found), addTimeout, TimeUnit.MILLISECONDS)) { + queueFullReported = false; logger.debug("Queued {} entries", evidenceCollection.size()); // If the collection has enough entries then start // taking data from it to be sent. @@ -677,11 +811,17 @@ private void processData(FlowData data, Map found) { trySendData(); } } else { - if (++lostData >= lostDataReset) { - logger.warn("Could not queue data for Share Usage. This could mean that" + - " the queue is too small, that the proportion of data shared is" + - " too high - or that the sending process has encountered" + - " problems sending."); + long lost = ++lostData; + if (queueFullReported == false || + lost % Constants.QUEUE_FULL_LOG_EVERY == 0) { + queueFullReported = true; + logger.warn("The usage sharing queue is full, so usage data is " + + "being discarded. This could mean that the queue is too " + + "small, that the proportion of data shared is too high, " + + "or that sending is failing. {} entries have been " + + "discarded.", lost); + } + if (lost >= lostDataReset) { lostData = 0; } } @@ -711,7 +851,9 @@ private ShareUsageData getDataFromEvidence(Map found) { continue; } - switch (entry.getKey()) { + // The keys are compared without regard to case, and the constants + // are lower case. + switch (entry.getKey().toLowerCase(Locale.ROOT)) { case fiftyone.pipeline.core.Constants.EVIDENCE_CLIENTIP_KEY: // The client IP is dealt with separately for backwards // compatibility purposes. @@ -789,6 +931,51 @@ private void submitSendData() { protected abstract void sendUsageData() throws HttpException; + /** + * Waits until sending may start again after failed sends. A consumer + * calls this before each message it sends. + * @return false if the element started closing while sending was paused, + * in which case nothing more should be sent + */ + protected boolean waitWhilePaused() { + long pause; + while ((pause = backOff.remaining(ShareUsageBackOff.now())) > 0) { + try { + if (closing.await(pause, TimeUnit.MILLISECONDS)) { + return false; + } + } catch (InterruptedException e) { + // Not re-interrupting, for the reason given in + // managedResourcesCleanup. The interrupt comes from closing. + return false; + } + } + return true; + } + + /** + * Records whether the remote service accepted a message, so that repeated + * failures pause sending and a success ends the pause. + * @param accepted true if the message was accepted + */ + protected void recordSendResult(boolean accepted) { + if (accepted) { + if (backOff.succeeded()) { + logger.info("Usage sharing resumed after a send succeeded."); + } + } else { + long pause = backOff.failed(ShareUsageBackOff.now()); + if (pause > 0) { + logger.warn( + "Usage sharing paused for {} seconds after failed " + + "sends. Usage data is still queued, up to the " + + "maximum queue size, and sending starts again when " + + "the pause ends.", + pause / 1000.0); + } + } + } + /** * Get engine Version No. * @return engine version no diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java index 56a96e9ac..c73a86326 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java @@ -129,7 +129,14 @@ public ShareUsageElement build() throws IOException { includedQueryStringParameters, ignoreDataEvidenceFilter, sessionCookieName, - null); + null, + failuresToEnterRecovery, + failuresWindowSeconds, + initialRecoveryDelaySeconds, + maxRecoveryDelaySeconds, + recoveryMultiplier, + shareAllEvidence, + shareAllQueryStringParameters); if (Objects.nonNull(this.httpClient)) { sue.httpClient = httpClient; } else if (Objects.nonNull(dataUploader)){ diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java index e376660cf..6e07f6dd3 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java @@ -59,6 +59,13 @@ public abstract class ShareUsageBuilderBase { protected final List includedQueryStringParameters = new ArrayList<>(); protected final List> ignoreDataEvidenceFilter = new ArrayList<>(); protected boolean trackSession; + protected int failuresToEnterRecovery = Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT; + protected int failuresWindowSeconds = Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT; + protected double initialRecoveryDelaySeconds = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT; + protected double maxRecoveryDelaySeconds = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT; + protected double recoveryMultiplier = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT; + protected boolean shareAllEvidence; + protected boolean shareAllQueryStringParameters; /** * Constructor @@ -143,6 +150,56 @@ public ShareUsageBuilderBase setIncludedQueryStringParameter( return this; } + /** + * If set to true, every query string and HTTP form parameter is shared, + * rather than only those that start with 51D_ or are named in the + * included parameters. The whole query string and the parameters that are + * never shared are still not shared. + *

+ * Default value is false + * @param shareAll true to share all query string parameters + * @return this builder + */ + @DefaultValue(booleanValue = false) + public ShareUsageBuilderBase setShareAllQueryStringParameters( + boolean shareAll) { + this.shareAllQueryStringParameters = shareAll; + return this; + } + + /** + * Whether every query string parameter is shared. + * @return true if all query string parameters are shared + */ + public boolean isShareAllQueryStringParameters() { + return shareAllQueryStringParameters; + } + + /** + * If set to true, all Evidence is shared, which overrides the blocked + * HTTP headers and the rules for cookies, session values and query string + * parameters. The keys that are never shared (an email address, the salt + * for hashed email identifiers, the whole query string, the path and the + * Cookie header) are still not shared. + *

+ * Default value is false + * @param shareAll true to share all Evidence + * @return this builder + */ + @DefaultValue(booleanValue = false) + public ShareUsageBuilderBase setShareAllEvidence(boolean shareAll) { + this.shareAllEvidence = shareAll; + return this; + } + + /** + * Whether all Evidence is shared. + * @return true if all Evidence is shared + */ + public boolean isShareAllEvidence() { + return shareAllEvidence; + } + /** * By default, all HTTP headers (excluding a few such as 'cookies', if they * don't start with 51D_) are shared. Individual headers can be excluded from sharing by @@ -420,6 +477,92 @@ public ShareUsageBuilderBase setTrackSession(boolean track) { return this; } + /** + * Set the number of failed sends within the failures window that pauses + * sending. Usage data is still queued while sending is paused, up to the + * maximum queue size. + * @param failuresToEnterRecovery the number of failed sends, from 1 to 100 + * @return this builder + */ + @DefaultValue(intValue = Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT) + public ShareUsageBuilderBase setFailuresToEnterRecovery( + int failuresToEnterRecovery) { + if (failuresToEnterRecovery < Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN || + failuresToEnterRecovery > Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX) { + throw new IllegalArgumentException( + "failuresToEnterRecovery must be from " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN + " to " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX + + ". Received: " + failuresToEnterRecovery + "."); + } + this.failuresToEnterRecovery = failuresToEnterRecovery; + return this; + } + + /** + * Set the time in seconds over which failed sends are counted. + * @param failuresWindowSeconds the time in seconds, 1 or more + * @return this builder + */ + @DefaultValue(intValue = Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT) + public ShareUsageBuilderBase setFailuresWindowSeconds( + int failuresWindowSeconds) { + if (failuresWindowSeconds < 1) { + throw new IllegalArgumentException( + "failuresWindowSeconds must be 1 or more."); + } + this.failuresWindowSeconds = failuresWindowSeconds; + return this; + } + + /** + * Set the length in seconds of the first pause after failed sends. + * @param initialDelaySeconds the length in seconds, more than 0 + * @return this builder + */ + @DefaultValue(doubleValue = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT) + public ShareUsageBuilderBase setInitialRecoveryDelaySeconds( + double initialDelaySeconds) { + if ((initialDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "initialDelaySeconds must be more than 0."); + } + this.initialRecoveryDelaySeconds = initialDelaySeconds; + return this; + } + + /** + * Set the longest a pause can be, in seconds. + * @param maxDelaySeconds the length in seconds, more than 0 + * @return this builder + */ + @DefaultValue(doubleValue = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT) + public ShareUsageBuilderBase setMaxRecoveryDelaySeconds( + double maxDelaySeconds) { + if ((maxDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "maxDelaySeconds must be more than 0."); + } + this.maxRecoveryDelaySeconds = maxDelaySeconds; + return this; + } + + /** + * Set what the pause is multiplied by each time the send tried when a + * pause ends fails. 2 doubles the pause each time. + * @param multiplier the multiplier, 1 or more + * @return this builder + */ + @DefaultValue(doubleValue = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT) + public ShareUsageBuilderBase setRecoveryMultiplier(double multiplier) { + if ((multiplier >= 1) == false) { + throw new IllegalArgumentException( + "multiplier must be 1 or more."); + } + this.recoveryMultiplier = multiplier; + return this; + } + /** * Create the {@link ShareUsageElement}. * @return the newly created target instance diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java index c56e032f8..37a5eba7c 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java @@ -46,6 +46,11 @@ import java.util.zip.GZIPOutputStream; import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_DEFAULT_HTTP_POST_TIMEOUT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT; import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_MAX_EVIDENCE_LENGTH; /** @@ -309,6 +314,98 @@ public class ShareUsageElement extends ShareUsageBase { List> ignoreDataEvidenceFilter, String sessionCookieName, Tracker tracker) { + this( + logger, + sharePercentage, + minimumEntriesPerMessage, + maximumQueueSize, + addTimeout, + takeTimeout, + repeatEvidenceIntervalMinutes, + trackSession, + shareUsageUrl, + blockedHttpHeaders, + includedQueryStringParameters, + ignoreDataEvidenceFilter, + sessionCookieName, + tracker, + SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT, + SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT, + SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT, + SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT, + SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT, + false, + false); + } + + /** + * Constructor + * + * @param logger the logger to use + * @param sharePercentage the approximate proportion of requests to share. + * 1 = 100%, 0.5 = 50%, etc. + * @param minimumEntriesPerMessage the minimum number of request entries per + * message sent to 51Degrees + * @param maximumQueueSize the maximum number of items to hold in the queue + * at one time. This must be larger than minimum + * entries + * @param addTimeout the timeout in milliseconds to allow when attempting to + * add an item to the queue. If this timeout is exceeded + * then the item is not shared + * @param takeTimeout the timeout in milliseconds to allow when attempting + * to take an item to the queue + * @param repeatEvidenceIntervalMinutes the interval (in minutes) which is + * used to decide if repeat evidence is + * old enough to consider a new session + * @param trackSession set if the tracker should consider sessions in share + * usage + * @param shareUsageUrl the URL to send data to + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParameters a list of the names of query string + * parameters that share usage should + * send to 51Degrees + * @param ignoreDataEvidenceFilter the filter used to determine if an item + * of evidence should be ignored or not + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param tracker the {@link Tracker} to use to determine if a given + * {@link FlowData} instance should be shared or not + * @param failuresToEnterRecovery the number of failed sends within the failures + * window that pauses sending + * @param failuresWindowSeconds the time in seconds over which failed sends are + * counted + * @param initialRecoveryDelaySeconds the length in seconds of the first pause + * @param maxRecoveryDelaySeconds the longest a pause can be, in seconds + * @param recoveryMultiplier what the pause is multiplied by each time the send + * tried when a pause ends fails + * @param shareAllEvidence if true then all Evidence is shared, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are shared + */ + ShareUsageElement( + Logger logger, + double sharePercentage, + int minimumEntriesPerMessage, + int maximumQueueSize, + int addTimeout, + int takeTimeout, + int repeatEvidenceIntervalMinutes, + boolean trackSession, + String shareUsageUrl, + List blockedHttpHeaders, + List includedQueryStringParameters, + List> ignoreDataEvidenceFilter, + String sessionCookieName, + Tracker tracker, + int failuresToEnterRecovery, + int failuresWindowSeconds, + double initialRecoveryDelaySeconds, + double maxRecoveryDelaySeconds, + double recoveryMultiplier, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { super( logger, sharePercentage, @@ -323,7 +420,14 @@ public class ShareUsageElement extends ShareUsageBase { includedQueryStringParameters, ignoreDataEvidenceFilter, sessionCookieName, - tracker); + tracker, + failuresToEnterRecovery, + failuresWindowSeconds, + initialRecoveryDelaySeconds, + maxRecoveryDelaySeconds, + recoveryMultiplier, + shareAllEvidence, + shareAllQueryStringParameters); headers.put("Content-Type", "text/xml; charset=utf-8"); // assume default GZIP connection @@ -336,7 +440,8 @@ public class ShareUsageElement extends ShareUsageBase { } /** - * Take data from the queue until there are fewer entries than the minimum batch size + * Take data from the queue until there are fewer entries than the minimum + * batch size, waiting out any pause after failed sends before each message */ @Override protected void sendUsageData() { @@ -345,10 +450,14 @@ protected void sendUsageData() { List allData = new ArrayList<>(); // drain queue while minimum entries in it do { + // Closing while paused ends the wait, and nothing more is sent. + if (waitWhilePaused() == false) { + break; + } logger.debug(threadMarker, "Queue size is {}", evidenceCollection.size()); try { allData.clear(); - // for historical reasons max entries set to 2* + // A message holds at most twice the minimum entries. while (allData.size() < minEntriesPerMessage * 2) { ShareUsageData currentData = evidenceCollection.poll(takeTimeout, TimeUnit.MILLISECONDS); // no data available @@ -362,7 +471,13 @@ protected void sendUsageData() { // drained everything - and there is no point posting a document // with no entries in it. if (allData.isEmpty() == false) { - sendAsXML(allData); + boolean accepted = false; + try { + sendAsXML(allData); + accepted = true; + } finally { + recordSendResult(accepted); + } } } catch (InterruptedException interruptedException) { // Deliberately not re-interrupting: the interrupt comes from @@ -394,7 +509,10 @@ protected void sendAsXML(List allData) throws Exception { // check for completion int code = dataUploader.getResponseCode(); if (code != 200) { - throw new Exception(String.format("Share Usage response code was %d ", code)); + throw new Exception(String.format( + "Share Usage response code was %d with content '%s'", + code, + dataUploader.getResponseContent())); } logger.debug(threadMarker, "Send elements done"); } @@ -555,6 +673,9 @@ protected void legacySendAsXML(List allData) throws Exception { } HttpURLConnection connection = httpClient.connect(new URI(shareUsageUrl.trim()).toURL()); + // A send that is never answered must end, or it stops all sending. + connection.setConnectTimeout(httpSendTimeout); + connection.setReadTimeout(httpSendTimeout); String response = httpClient.postData(connection, headers, baos.toByteArray()); int responseCode = connection.getResponseCode(); String responseMessage = connection.getResponseMessage() + "data: '" + response + "'"; diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java index c6082300e..5c8f2e478 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java @@ -22,7 +22,6 @@ package fiftyone.pipeline.engines.fiftyone.trackers; -import fiftyone.caching.PutCache; import fiftyone.pipeline.core.data.DataKey; import fiftyone.pipeline.core.data.DataKeyBuilder; import fiftyone.pipeline.core.data.DataKeyBuilderDefault; @@ -32,7 +31,6 @@ import fiftyone.pipeline.engines.fiftyone.data.ShareUsageEvidence; import fiftyone.pipeline.engines.trackers.TrackerBase; -import java.io.IOException; import java.util.Date; import java.util.Map; @@ -53,13 +51,6 @@ public class ShareUsageTracker extends TrackerBase { */ private final long interval; - /** - * The times keyed by the values usage sharing would share. The cache of - * the base class is keyed by the Evidence alone, which no longer holds - * the web request's values. - */ - private final PutCache sharedValuesCache; - /** * Construct a new instance. * @param configuration the cache configuration to use when building the @@ -75,10 +66,6 @@ public ShareUsageTracker( super(configuration); this.interval = intervalMillis; this.filter = filter; - // The base class has checked that the builder makes a PutCache. - this.sharedValuesCache = (PutCache) - configuration.getCacheBuilder().build( - null, configuration.getSize()); } @Override @@ -98,31 +85,14 @@ protected EvidenceKeyFilter getFilter() { return filter; } - @Override - public Date get(FlowData data) { - return sharedValuesCache.get(getSharedValuesKey(data)); - } - - @Override - public void put(FlowData data, Date value) { - sharedValuesCache.put(getSharedValuesKey(data), value); - } - - @Override - public void close() throws IOException { - try { - sharedValuesCache.close(); - } finally { - super.close(); - } - } - /** - * The key of the values {@link ShareUsageEvidence#find} finds that the - * filter includes, which take in the web request's values, as the - * session tracking hash is of the values that would be shared. + * The key is made from the values {@link ShareUsageEvidence#find} finds + * that the filter includes, which take in the web request's values, + * as the session tracking hash is of the values that would be shared. + * The Evidence alone does not hold them. */ - private DataKey getSharedValuesKey(FlowData data) { + @Override + protected DataKey generateKey(FlowData data) { DataKeyBuilder builder = new DataKeyBuilderDefault(); for (Map.Entry entry : ShareUsageEvidence.find(data.getEvidence()).entrySet()) { diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java new file mode 100644 index 000000000..1a628235c --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java @@ -0,0 +1,199 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The rules that decide when usage sharing may send after failed sends, + * with each test setting the time, and the builder settings for them. + */ +public class ShareUsageBackOffTests { + + private static final long START = 1767225600000L; + + private static long at(double seconds) { + return START + Math.round(seconds * 1000); + } + + /** + * Three failures within a ten second window start a two second pause, + * rising by two each time up to five. + */ + private static ShareUsageBackOff create( + int failuresToEnterRecovery, + double initialDelaySeconds, + double maxDelaySeconds) { + return new ShareUsageBackOff( + failuresToEnterRecovery, + 10, + initialDelaySeconds, + maxDelaySeconds, + 2); + } + + private static ShareUsageBackOff create() { + return create(3, 2, 5); + } + + /** + * The time the back-off is given comes from a clock that a change to the + * system time does not move. + */ + @Test + public void ShareUsageBackOff_Now_FollowsTheMonotonicClock() { + long before = System.nanoTime() / 1000000; + long now = ShareUsageBackOff.now(); + long after = System.nanoTime() / 1000000; + + assertTrue(before <= now && now <= after); + } + + @Test + public void ShareUsageBackOff_PausesAfterFailuresWithinWindow() { + ShareUsageBackOff backOff = create(); + + assertEquals(0, backOff.failed(at(0))); + assertEquals(0, backOff.failed(at(1))); + assertEquals(0, backOff.remaining(at(1)), + "Sending paused before enough sends failed."); + assertFalse(backOff.isBackingOff()); + + assertEquals(2000, backOff.failed(at(2))); + assertTrue(backOff.isBackingOff()); + assertEquals(1000, backOff.remaining(at(3))); + assertEquals(0, backOff.remaining(at(4))); + } + + @Test + public void ShareUsageBackOff_FailuresOutsideWindowDoNotCount() { + ShareUsageBackOff backOff = create(); + + backOff.failed(at(0)); + backOff.failed(at(1)); + + assertEquals(0, backOff.failed(at(11.5)), + "Failures older than the window were counted."); + assertFalse(backOff.isBackingOff()); + } + + /** + * The window counts failures, not failures in a row, so a high failure + * rate pauses sending even when some sends succeed. + */ + @Test + public void ShareUsageBackOff_SuccessesBetweenFailuresDoNotReset() { + ShareUsageBackOff backOff = create(); + + backOff.failed(at(0)); + assertFalse(backOff.succeeded()); + backOff.failed(at(1)); + assertFalse(backOff.succeeded()); + + assertEquals(2000, backOff.failed(at(2))); + } + + @Test + public void ShareUsageBackOff_FailedSendAfterPauseMultipliesDelay() { + ShareUsageBackOff backOff = create(1, 2, 5); + + assertEquals(2000, backOff.failed(at(0))); + assertEquals(4000, backOff.failed(at(2))); + assertEquals(5000, backOff.failed(at(6)), + "The delay grew past the maximum."); + assertEquals(5000, backOff.failed(at(11))); + assertEquals(5000, backOff.remaining(at(11))); + } + + @Test + public void ShareUsageBackOff_SuccessEndsBackOff() { + ShareUsageBackOff backOff = create(2, 2, 5); + backOff.failed(at(0)); + backOff.failed(at(1)); + backOff.failed(at(3)); + + assertTrue(backOff.succeeded(), + "A success did not end the back-off."); + assertFalse(backOff.isBackingOff()); + assertEquals(0, backOff.remaining(at(4))); + + // The next pause needs the full number of failures again, and starts + // from the initial delay. + assertEquals(0, backOff.failed(at(8))); + assertEquals(2000, backOff.failed(at(9))); + } + + @Test + public void ShareUsageBackOff_MaximumBelowInitialLimitsFirstPause() { + ShareUsageBackOff backOff = create(1, 10, 3); + + assertEquals(3000, backOff.failed(at(0))); + } + + @Test + public void ShareUsageBackOff_InvalidSettingsAreRejected() { + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(0, 10, 2, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(101, 10, 2, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 0, 2, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, 0, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, Double.NaN, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, 2, 0, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, 2, 5, 0.5)); + } + + /** + * The builder settings that pause sending after failed sends reject + * values the pause cannot work with. + */ + @Test + public void ShareUsageBuilder_RecoverySettings_RejectValuesOutOfRange() { + ShareUsageBuilder builder = + new ShareUsageBuilder(LoggerFactory.getILoggerFactory()); + + assertThrows(IllegalArgumentException.class, + () -> builder.setFailuresToEnterRecovery(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setFailuresToEnterRecovery(101)); + assertThrows(IllegalArgumentException.class, + () -> builder.setFailuresWindowSeconds(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setInitialRecoveryDelaySeconds(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setMaxRecoveryDelaySeconds(-1)); + assertThrows(IllegalArgumentException.class, + () -> builder.setRecoveryMultiplier(0.5)); + } +} diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java new file mode 100644 index 000000000..0a6405558 --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java @@ -0,0 +1,436 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import fiftyone.pipeline.core.data.FlowData; +import fiftyone.pipeline.core.flowelements.Pipeline; +import fiftyone.pipeline.core.flowelements.PipelineBuilder; +import fiftyone.pipeline.engines.services.DataUploader; +import fiftyone.pipeline.engines.trackers.Tracker; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.Map; +import java.util.function.BooleanSupplier; +import java.util.function.Supplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static fiftyone.pipeline.core.Constants.EVIDENCE_CLIENTIP_KEY; +import static fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +/** + * Usage sharing pauses sending after repeated failed sends, keeps queueing + * evidence while paused, and drains the queue once a send succeeds. These + * tests use the default settings, which pause sending for two seconds after + * ten sends fail within 100 seconds. + */ +public class ShareUsageElementBackOffTests { + + /** + * The longest a test waits for something it expects to happen. + */ + private static final long WAIT_LIMIT_MILLIS = 30000; + + private static final int DEFAULT_FAILURES = 10; + + private static final long DEFAULT_PAUSE_MILLIS = 2000; + + private static final Pattern CLIENT_IP = + Pattern.compile("([^<]*)"); + + /** + * Answers each send with the status the test sets, and records when each + * send was answered and the client IPs of the sends it accepted. + */ + private static class RecordingUploader implements DataUploader { + private final long start = System.nanoTime(); + private final List times = new ArrayList<>(); + private final List accepted = new ArrayList<>(); + private volatile int code = 500; + private volatile String content = ""; + private ByteArrayOutputStream body; + + @Override + public synchronized OutputStream getOutputStream() { + body = new ByteArrayOutputStream(); + return body; + } + + @Override + public synchronized int getResponseCode() { + int answer = code; + times.add(now()); + if (answer == 200) { + Matcher matcher = CLIENT_IP.matcher( + new String(body.toByteArray(), StandardCharsets.UTF_8)); + while (matcher.find()) { + accepted.add(matcher.group(1)); + } + } + return answer; + } + + public String getResponseContent() { + return content; + } + + long now() { + return (System.nanoTime() - start) / 1000000; + } + + synchronized int count() { + return times.size(); + } + + synchronized List times() { + return new ArrayList<>(times); + } + + synchronized List accepted() { + return new ArrayList<>(accepted); + } + } + + /** + * Shares every request. + */ + private static class ShareAll implements Tracker { + @Override + public boolean track(FlowData flowData) { + return true; + } + } + + private Pipeline pipeline; + private Logger logger = LoggerFactory.getLogger(ShareUsageElement.class); + private RecordingUploader uploader; + private ShareUsageElement element; + private int requests; + + /** + * Each entry is sent on its own, as the minimum entries per message is + * one. + */ + @BeforeEach + public void init() throws Exception { + pipeline = new PipelineBuilder().build(); + uploader = new RecordingUploader(); + requests = 0; + element = new ShareUsageElement( + LoggerFactory.getLogger(ShareUsageElement.class), + 1, + 1, + 100, + 100, + 100, + 1, + true, + "http://localhost/usage", + new ArrayList<>(), + new ArrayList<>(), + new ArrayList<>(), + DEFAULT_SESSION_COOKIE_NAME, + new ShareAll()); + element.addPipeline(pipeline); + element.dataUploader = uploader; + } + + /** + * Replaces the element with one that has its own back-off settings. The + * failures window is long enough that every failure in a test counts. + */ + private void useBackOff( + int failuresToEnterRecovery, + double initialDelaySeconds, + double maxDelaySeconds, + int maximumQueueSize) throws Exception { + element.close(); + element = new ShareUsageElement( + logger, + 1, + 1, + maximumQueueSize, + 100, + 100, + 1, + true, + "http://localhost/usage", + new ArrayList<>(), + new ArrayList<>(), + new ArrayList<>(), + DEFAULT_SESSION_COOKIE_NAME, + new ShareAll(), + failuresToEnterRecovery, + 100, + initialDelaySeconds, + maxDelaySeconds, + 2, + false, + false); + element.addPipeline(pipeline); + element.dataUploader = uploader; + } + + @AfterEach + public void cleanUp() throws Exception { + element.close(); + pipeline.close(); + } + + /** + * Processes a request whose evidence no other request has, and returns + * its client IP. + */ + private String process() throws Exception { + int number = requests++; + String ip = "10.0." + (number / 250) + "." + (number % 250 + 1); + Map evidence = new HashMap<>(); + evidence.put(EVIDENCE_CLIENTIP_KEY, ip); + try (FlowData data = pipeline.createFlowData()) { + data.addEvidence(evidence); + element.process(data); + } + return ip; + } + + private static void waitFor( + BooleanSupplier condition, + Supplier message) throws InterruptedException { + long end = System.currentTimeMillis() + WAIT_LIMIT_MILLIS; + while (condition.getAsBoolean() == false) { + if (System.currentTimeMillis() > end) { + fail(message.get()); + } + Thread.sleep(20); + } + } + + /** + * Processes requests one at a time, waiting for each to be sent, so each + * is a separate failed send. + */ + private void failSends(int count) throws Exception { + for (int i = 1; i <= count; i++) { + process(); + final int expected = i; + waitFor( + () -> uploader.count() == expected && + element.isRunning() == false, + () -> "Send " + expected + " was not made."); + } + } + + /** + * Ten failed sends pause sending. Nothing is sent during the pause, the + * evidence processed during it is queued, and all of it is sent once the + * pause ends and the service answers, with no more requests processed to + * start the sending. + */ + @Test + public void ShareUsageElement_BackOff_PausesThenDrainsQueue() + throws Exception { + failSends(DEFAULT_FAILURES); + long pausedAt = uploader.now(); + + List queued = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + queued.add(process()); + } + Thread.sleep(700); + assertEquals(DEFAULT_FAILURES, uploader.count(), + "Sends were made during the pause that ten failed sends should " + + "have started."); + + uploader.code = 200; + waitFor( + () -> uploader.accepted().containsAll(queued), + () -> "The evidence processed during the pause was not all " + + "sent once the service answered. Sent: " + uploader.accepted()); + long waited = uploader.times().get(DEFAULT_FAILURES) - pausedAt; + assertTrue(waited >= DEFAULT_PAUSE_MILLIS - 500, + "Sending resumed " + waited + " ms after the pause started, " + + "before the two second pause ended."); + } + + /** + * Closing while paused returns at once rather than waiting for the pause + * to end, and sends nothing. + */ + @Test + public void ShareUsageElement_BackOff_CloseWhilePausedDoesNotWaitOrSend() + throws Exception { + failSends(DEFAULT_FAILURES); + process(); + process(); + + long start = System.currentTimeMillis(); + element.close(); + long took = System.currentTimeMillis() - start; + + assertTrue(took < 1500, + "Closing took " + took + " ms, as it waited for the pause."); + assertEquals(DEFAULT_FAILURES, uploader.count(), + "A send was made during the pause or when closing."); + } + + /** + * Each failed send after a pause multiplies the delay, and the delay + * stops growing at the maximum. + */ + @Test + public void ShareUsageElement_BackOff_DelayGrowsToMaximum() + throws Exception { + useBackOff(1, 0.5, 1.0, 100); + for (int i = 0; i < 40; i++) { + process(); + } + + waitFor( + () -> uploader.count() >= 4, + () -> "Fewer than four sends were made."); + List times = uploader.times(); + long first = times.get(1) - times.get(0); + long second = times.get(2) - times.get(1); + long third = times.get(3) - times.get(2); + String shown = first + ", " + second + ", " + third + " ms"; + assertTrue(first >= 400, + "The first pause was not the initial delay. Gaps: " + shown); + assertTrue(second >= 900, + "The second pause was not doubled. Gaps: " + shown); + assertTrue(third >= 900 && third < 1900, + "The third pause was not held at the maximum. Gaps: " + shown); + } + + /** + * When the queue fills during a pause, new evidence is dropped and what + * was queued first is sent once sending resumes. + */ + @Test + public void ShareUsageElement_BackOff_FullQueueDropsNewEvidence() + throws Exception { + useBackOff(1, 2, 2, 4); + failSends(1); + + List kept = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + kept.add(process()); + } + List dropped = new ArrayList<>(); + for (int i = 0; i < 6; i++) { + dropped.add(process()); + } + uploader.code = 200; + + waitFor( + () -> uploader.accepted().containsAll(kept), + () -> "The evidence queued during the pause was not all sent. " + + "Sent: " + uploader.accepted()); + Thread.sleep(500); + for (String ip : dropped) { + assertTrue(uploader.accepted().contains(ip) == false, + "Evidence processed when the queue was full was sent."); + } + } + + /** + * A logger that records the warnings it is given. + */ + private static Logger recordingLogger(List warnings) { + return (Logger) Proxy.newProxyInstance( + Logger.class.getClassLoader(), + new Class[] {Logger.class}, + (proxy, method, args) -> { + if (method.getName().equals("warn")) { + warnings.add(Arrays.deepToString(args)); + return null; + } + if (method.getReturnType() == boolean.class) { + return false; + } + if (method.getReturnType() == int.class) { + return 0; + } + if (method.getReturnType() == String.class) { + return "test"; + } + return null; + }); + } + + /** + * Evidence discarded because the queue is full is logged, but not with a + * message for each entry discarded. + */ + @Test + public void ShareUsageElement_FullQueue_LogsWithoutAMessagePerEntry() + throws Exception { + List warnings = new CopyOnWriteArrayList<>(); + logger = recordingLogger(warnings); + useBackOff(1, 2, 2, 4); + failSends(1); + warnings.clear(); + + for (int i = 0; i < 10; i++) { + process(); + } + + long full = warnings.stream() + .filter(warning -> warning.contains("full")) + .count(); + assertTrue(full >= 1, "Discarding entries was not logged."); + assertTrue(full < 6, "A message was logged for each entry discarded."); + } + + /** + * A send answered with anything other than 200 reports the status code + * and the content of the answer. + */ + @Test + public void ShareUsageElement_FailedAnswer_ReportsStatusAndContent() { + uploader.content = "the service says no"; + + Exception failure = assertThrows( + Exception.class, + () -> element.sendAsXML(new ArrayList<>())); + + assertTrue(failure.getMessage().contains("500")); + assertTrue(failure.getMessage().contains("the service says no")); + } +} diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java new file mode 100644 index 000000000..65a920482 --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java @@ -0,0 +1,282 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import fiftyone.caching.Cache; +import fiftyone.caching.CacheBuilder; +import fiftyone.caching.LruPutCache; +import fiftyone.pipeline.core.configuration.ElementOptions; +import fiftyone.pipeline.core.configuration.PipelineOptions; +import fiftyone.pipeline.core.data.FlowData; +import fiftyone.pipeline.core.flowelements.Pipeline; +import fiftyone.pipeline.core.flowelements.PipelineBuilder; +import fiftyone.pipeline.engines.configuration.CacheConfiguration; +import fiftyone.pipeline.engines.fiftyone.data.EvidenceKeyFilterShareUsage; +import fiftyone.pipeline.engines.fiftyone.trackers.ShareUsageTracker; +import fiftyone.pipeline.engines.services.DataUploader; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.w3c.dom.Document; +import org.xml.sax.InputSource; + +import javax.xml.parsers.DocumentBuilderFactory; +import javax.xml.xpath.XPathFactory; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the usage sharing options "Share all Evidence" and "Share all query + * string parameters" from a configuration file, the case of the keys the + * element reads and the number of caches a tracker holds. + * @see Specification + */ +public class ShareUsageOptionsTests { + + /** + * Keeps what the element sends so a test can read it. + */ + private static class CapturingUploader implements DataUploader { + private final ByteArrayOutputStream sent = new ByteArrayOutputStream(); + + @Override + public OutputStream getOutputStream() { + return sent; + } + + @Override + public int getResponseCode() { + return 200; + } + + Document document() throws Exception { + return DocumentBuilderFactory.newInstance() + .newDocumentBuilder() + .parse(new InputSource( + new ByteArrayInputStream(sent.toByteArray()))); + } + } + + /** + * Counts the caches the tracker asks for. + */ + private static class CountingCacheBuilder implements CacheBuilder { + final AtomicInteger built = new AtomicInteger(); + private final LruPutCache.Builder inner = new LruPutCache.Builder(); + + @Override + public Cache build(Cache c, int cacheSize) { + built.incrementAndGet(); + return inner.build(c, cacheSize); + } + } + + /** + * Options for a usage sharing element that sends every request at once, + * as a configuration file holds them. + */ + private static PipelineOptions options(String... nameAndValue) { + ElementOptions element = new ElementOptions(); + element.builderName = "ShareUsage"; + element.buildParameters.put("SharePercentage", "1"); + element.buildParameters.put("MinimumEntriesPerMessage", "1"); + element.buildParameters.put("RepeatEvidenceIntervalMinutes", "0"); + element.buildParameters.put("BlockedHttpHeader", "x-blocked"); + for (int i = 0; i < nameAndValue.length; i += 2) { + element.buildParameters.put(nameAndValue[i], nameAndValue[i + 1]); + } + PipelineOptions options = new PipelineOptions(); + options.elements.add(element); + return options; + } + + private static Map evidence() { + Map evidence = new HashMap<>(); + evidence.put("header.user-agent", "agent"); + evidence.put("header.x-blocked", "blocked"); + evidence.put("header.cookie", "a=b"); + evidence.put("cookie.plain", "cookie"); + evidence.put("session.plain", "session"); + evidence.put("query.plain", "query"); + evidence.put("query.id.email", "a@example.com"); + evidence.put("id.email", "a@example.com"); + evidence.put("id.salt", "salt"); + evidence.put("server.request-path", "/path"); + evidence.put("server.request-query", "?plain=query"); + return evidence; + } + + /** + * Builds the element from the options, gives it the Evidence and returns + * what it sends. + */ + private static Document send( + PipelineOptions options, + Map evidence) throws Exception { + CapturingUploader uploader = new CapturingUploader(); + try (Pipeline pipeline = new PipelineBuilder( + LoggerFactory.getILoggerFactory()) + .buildFromConfiguration(options)) { + ShareUsageElement element = + pipeline.getElement(ShareUsageElement.class); + assertNotNull(element); + element.dataUploader = uploader; + try (FlowData data = pipeline.createFlowData()) { + data.addEvidence(evidence); + element.process(data); + assertNotNull(element.getSendDataFuture()); + element.getSendDataFuture().get(); + } + } + return uploader.document(); + } + + private static boolean shared( + Document sent, + String category, + String name) throws Exception { + return !XPathFactory.newInstance().newXPath().evaluate( + "//Device/" + category + "[@Name='" + name + "']", + sent).isEmpty(); + } + + @Test + public void defaults_ShareOnlyWhatTheRulesAllow() throws Exception { + Document sent = send(options(), evidence()); + + assertTrue(shared(sent, "header", "user-agent")); + assertFalse(shared(sent, "header", "x-blocked")); + assertFalse(shared(sent, "cookie", "plain")); + assertFalse(shared(sent, "session", "plain")); + assertFalse(shared(sent, "query", "plain")); + } + + @Test + public void shareAllEvidence_FromConfiguration_SharesEverythingButNeverShared() + throws Exception { + Document sent = send(options("ShareAllEvidence", "true"), evidence()); + + assertTrue(shared(sent, "header", "user-agent")); + assertTrue(shared(sent, "header", "x-blocked")); + assertTrue(shared(sent, "cookie", "plain")); + assertTrue(shared(sent, "session", "plain")); + assertTrue(shared(sent, "query", "plain")); + assertFalse(shared(sent, "query", "id.email")); + assertFalse(shared(sent, "id", "email")); + assertFalse(shared(sent, "id", "salt")); + assertFalse(shared(sent, "header", "cookie")); + assertFalse(shared(sent, "server", "request-path")); + assertFalse(shared(sent, "server", "request-query")); + } + + @Test + public void shareAllEvidence_False_FromConfiguration_KeepsTheRules() + throws Exception { + Document sent = send(options("ShareAllEvidence", "false"), evidence()); + + assertFalse(shared(sent, "header", "x-blocked")); + assertFalse(shared(sent, "cookie", "plain")); + assertFalse(shared(sent, "query", "plain")); + } + + @Test + public void shareAllQueryStringParameters_FromConfiguration_SharesEveryParameter() + throws Exception { + Document sent = send( + options("ShareAllQueryStringParameters", "true"), + evidence()); + + assertTrue(shared(sent, "query", "plain")); + assertFalse(shared(sent, "query", "id.email")); + assertFalse(shared(sent, "server", "request-query")); + // The other rules still apply. + assertFalse(shared(sent, "header", "x-blocked")); + assertFalse(shared(sent, "cookie", "plain")); + assertFalse(shared(sent, "session", "plain")); + assertTrue(shared(sent, "header", "user-agent")); + } + + @Test + public void shareAllQueryStringParameters_False_FromConfiguration_KeepsTheRules() + throws Exception { + Document sent = send( + options("ShareAllQueryStringParameters", "false"), + evidence()); + + assertFalse(shared(sent, "query", "plain")); + } + + @Test + public void builder_ShareOptionsDefaultToFalse() { + ShareUsageBuilder builder = new ShareUsageBuilder( + LoggerFactory.getILoggerFactory()); + + assertFalse(builder.isShareAllEvidence()); + assertFalse(builder.isShareAllQueryStringParameters()); + } + + @Test + public void keysAreReadWithoutRegardToCase() throws Exception { + Map evidence = new HashMap<>(); + evidence.put("Server.Client-IP", "1.2.3.4"); + evidence.put("Query.Session-ID", "session-1"); + evidence.put("Query.Sequence", "5"); + + Document sent = send(options(), evidence); + + assertEquals("1.2.3.4", + sent.getElementsByTagName("ClientIP").item(0).getTextContent()); + assertEquals("session-1", + sent.getElementsByTagName("SessionId").item(0).getTextContent()); + assertEquals("5", + sent.getElementsByTagName("Sequence").item(0).getTextContent()); + } + + @Test + public void tracker_BuildsOneCache() { + CountingCacheBuilder builder = new CountingCacheBuilder(); + + try (ShareUsageTracker tracker = new ShareUsageTracker( + new CacheConfiguration(builder, 10), + 1000, + new EvidenceKeyFilterShareUsage( + new ArrayList(), + new ArrayList(), + false, + DEFAULT_SESSION_COOKIE_NAME))) { + assertEquals(1, builder.built.get()); + } catch (java.io.IOException e) { + throw new IllegalStateException(e); + } + } +} diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java new file mode 100644 index 000000000..780cb80bb --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java @@ -0,0 +1,133 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import fiftyone.pipeline.engines.services.HttpClient; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import java.net.HttpURLConnection; +import java.net.URL; +import java.util.ArrayList; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A send through an {@link HttpClient} that never answers must end, or it + * would stop all sending. + */ +public class ShareUsageSendTimeoutTests { + + /** + * Opens a connection that records the timeouts set on it. + */ + private static class RecordingHttpClient implements HttpClient { + private HttpURLConnection connection; + + @Override + public HttpURLConnection connect(URL url) { + connection = new HttpURLConnection(url) { + @Override + public int getResponseCode() { + return 200; + } + + @Override + public String getResponseMessage() { + return ""; + } + + @Override + public void disconnect() { } + + @Override + public boolean usingProxy() { + return false; + } + + @Override + public void connect() { } + }; + return connection; + } + + @Override + public String postData( + HttpURLConnection connection, + Map headers, + byte[] data) { + return ""; + } + + @Override + public String getResponseString(HttpURLConnection connection) { + return ""; + } + + @Override + public String getResponseString( + HttpURLConnection connection, + Map headers) { + return ""; + } + } + + @Test + public void ShareUsageElement_LegacySend_ConnectionHasTimeouts() + throws Exception { + RecordingHttpClient client = new RecordingHttpClient(); + ShareUsageElement element = new ShareUsageElement( + LoggerFactory.getLogger(ShareUsageElement.class), + 1, + 1, + 100, + 100, + 100, + 1, + true, + "http://localhost/usage", + new ArrayList<>(), + new ArrayList<>(), + new ArrayList<>(), + fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME, + flowData -> true); + element.httpClient = client; + try { + element.legacySendAsXML(new ArrayList<>()); + + assertEquals( + element.httpSendTimeout, + client.connection.getConnectTimeout(), + "The connection has no connect timeout."); + assertEquals( + element.httpSendTimeout, + client.connection.getReadTimeout(), + "The connection has no read timeout."); + assertTrue(element.httpSendTimeout > 0); + } finally { + element.close(); + } + } +} diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java index fd16b2c1b..348f613ac 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java @@ -70,14 +70,26 @@ public DataKeyedCacheBase(CacheConfiguration configuration) { */ protected abstract EvidenceKeyFilter getFilter(); + /** + * Creates the key the internal cache holds the data of a {@link FlowData} + * under. By default this is the key the {@link FlowData} generates from + * the Evidence that {@link #getFilter()} includes. A class whose key is + * made from other values overrides this method. + * @param flowData the {@link FlowData} to make a key for + * @return the key to use in the internal cache + */ + protected DataKey generateKey(FlowData flowData) { + return flowData.generateKey(getFilter()); + } + @Override public V get(FlowData flowData) { - return internalCache.get(flowData.generateKey(getFilter())); + return internalCache.get(generateKey(flowData)); } @Override public void put(FlowData flowData, V value) { - internalCache.put(flowData.generateKey(getFilter()), value); + internalCache.put(generateKey(flowData), value); } @Override diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java index 38df1b000..b790a6415 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java @@ -44,5 +44,12 @@ public interface DataUploader { */ int getResponseCode() throws IOException; + /** + * @return the content of the answer, which is logged when the code is + * not 200, or an empty string if the uploader cannot give it + */ + default String getResponseContent() { + return ""; + } } diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java index 777584bf5..9ad54e674 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java @@ -23,9 +23,11 @@ package fiftyone.pipeline.engines.services; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; import java.net.HttpURLConnection; import java.net.URI; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.zip.GZIPOutputStream; @@ -34,13 +36,19 @@ public class DataUploaderHttp implements DataUploader{ private final String url; private final Map headers; private final int timeout; + + /** + * The most characters of an answer's content that are given. + */ + private static final int LOGGED_CONTENT_LENGTH = 500; HttpURLConnection connection; /** * the URL to which data is to be uploaded * @param url the URL rto which data should be uploaded * @param headers HTTP headers for the connection - * @param timeout connection timeout in milliseconds + * @param timeout milliseconds to wait to connect, and then to wait for + * an answer */ public DataUploaderHttp(String url, Map headers, int timeout) { this.url = url; @@ -55,6 +63,7 @@ public DataUploaderHttp(String url, Map headers, int timeout) { public OutputStream getOutputStream() throws Exception{ connection = (HttpURLConnection) new URI(url.trim()).toURL().openConnection(); connection.setConnectTimeout(timeout); + connection.setReadTimeout(timeout); connection.setRequestMethod("POST"); for (Map.Entry header : headers.entrySet()) { connection.setRequestProperty(header.getKey(), header.getValue()); @@ -67,5 +76,28 @@ public OutputStream getOutputStream() throws Exception{ public int getResponseCode() throws IOException { return connection.getResponseCode(); } + + /** + * Reads the content of a failed answer, as far as the most characters + * that are logged. + */ + @Override + public String getResponseContent() { + try (InputStream error = connection.getErrorStream()) { + if (error == null) { + return ""; + } + byte[] content = new byte[LOGGED_CONTENT_LENGTH]; + int length = 0; + int read; + while (length < content.length && (read = error.read( + content, length, content.length - length)) > 0) { + length += read; + } + return new String(content, 0, length, StandardCharsets.UTF_8); + } catch (IOException e) { + return ""; + } + } } diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java new file mode 100644 index 000000000..5948dfac1 --- /dev/null +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java @@ -0,0 +1,118 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.services; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import com.sun.net.httpserver.HttpServer; + +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.util.HashMap; + +import org.junit.Test; + +/** + * An upload to a service that accepts the connection and never answers must + * end, or it would stop all usage sharing. + */ +public class DataUploaderHttpTests { + + /** + * How long the uploader waits for an answer. + */ + private static final int TIMEOUT_MILLIS = 500; + + /** + * How long the test service holds a connection open without answering. + */ + private static final int HOLD_MILLIS = 30000; + + /** + * The test fails if the upload has not ended by then. + */ + private static final int LIMIT_MILLIS = 15000; + + @Test(timeout = LIMIT_MILLIS) + public void DataUploaderHttp_NoAnswer_TimesOut() throws Exception { + try (ServerSocket server = new ServerSocket(0)) { + Thread holder = new Thread(() -> { + try (Socket socket = server.accept()) { + Thread.sleep(HOLD_MILLIS); + } catch (Exception ignored) { + // The test has ended and closed the server. + } + }); + holder.setDaemon(true); + holder.start(); + DataUploaderHttp uploader = new DataUploaderHttp( + "http://localhost:" + server.getLocalPort() + "/usage", + new HashMap<>(), + TIMEOUT_MILLIS); + try (OutputStream out = uploader.getOutputStream()) { + out.write(new byte[] {1}); + } + + try { + uploader.getResponseCode(); + fail("The upload was answered."); + } catch (SocketTimeoutException expected) { + // The wait for an answer ended. + } + } + } + + @Test(timeout = LIMIT_MILLIS) + public void DataUploaderHttp_FailedAnswer_GivesTheContent() + throws Exception { + byte[] answer = "the service says no".getBytes(StandardCharsets.UTF_8); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext("/usage", exchange -> { + exchange.sendResponseHeaders(500, answer.length); + exchange.getResponseBody().write(answer); + exchange.close(); + }); + server.start(); + try { + DataUploaderHttp uploader = new DataUploaderHttp( + "http://localhost:" + server.getAddress().getPort() + "/usage", + new HashMap<>(), + TIMEOUT_MILLIS * 10); + try (OutputStream out = uploader.getOutputStream()) { + out.write(new byte[] {1}); + } + + assertEquals(500, uploader.getResponseCode()); + assertEquals( + "the service says no", + uploader.getResponseContent()); + } finally { + server.stop(0); + } + } +}