remove sleep loop from moth/setup routines.

sarracenia/__init__.py

@@ -417,13 +417,17 @@ def durationToSeconds(str_value, default=None) -> float:
     206: "Partial Content: received and inserted.",
     304: "Not modified (Checksum validated, unchanged, so no download resulted.)",
     307: "Insertion deferred (writing to temporary part file for the moment.)",
+    404: "Not Found: no pattern match",
+    406: "Not Acceptable: file older than fileAgeMax",
     410: "Gone: server data different from notification message",
     417: "Expectation Failed: invalid notification message (corrupt headers)",
     422: "Unprocessable Content: could not determine path to transfer to",
+    425: "Too Early: file younger than fileAgeMin",
     499: "Failure: Not Copied. SFTP/FTP/HTTP download problem",
     #FIXME : should  not have 503 error code 3 times in a row
     # 503: "Service unavailable. delete (File removal not currently supported.)",
     503: "Unable to process: Service unavailable",
+    504: "Gateway Timeout: message too old"
     # 503: "Unsupported transport protocol specified in posting."
 }
 

sarracenia/config/__init__.py

@@ -96,6 +96,7 @@ def __repr__(self) -> str:
     'inline': False,
     'inlineOnly': False,
     'identity_method': 'sha512',
+    'logDuplicates': False,
     'logFormat': '%(asctime)s [%(levelname)s] %(name)s %(funcName)s %(message)s',
     'logMetrics': False,
     'logStdout': False,
@@ -142,7 +143,7 @@ def __repr__(self) -> str:
 # all the boolean settings.
 flag_options = [ 'acceptSizeWrong', 'acceptUnmatched', 'amqp_consumer', 'baseUrl_relPath', 'debug', \
     'delete', 'discard', 'download', 'dry_run', 'durable', 'exchangeDeclare', 'exchangeSplit', 'logReject', 'realpathFilter', \
-    'follow_symlinks', 'force_polling', 'inline', 'inlineOnly', 'inplace', 'logMetrics', 'logStdout', 'logReject', 'restore', \
+    'follow_symlinks', 'force_polling', 'inline', 'inlineOnly', 'inplace', 'logDuplicates', 'logMetrics', 'logStdout', 'logReject', 'restore', \
     'messageDebugDump', 'mirror', 'timeCopy', 'notify_only', 'overwrite', 'post_on_start', \
     'permCopy', 'persistent', 'queueBind', 'queueDeclare', 'randomize', 'recursive', 'realpathPost', \
     'reconnect', 'report', 'reset', 'retry_refilter', 'retryEmptyBeforeExit', 'save', 
@@ -164,7 +165,7 @@ def __repr__(self) -> str:
 set_options = [ 'logEvents', 'fileEvents' ]
 
 set_choices = { 
-    'logEvents' : set(sarracenia.flowcb.entry_points + [ 'reject' ]),
+    'logEvents' : set(sarracenia.flowcb.entry_points + [ 'reject', 'nodupe' ]),
     'fileEvents' : set( [ 'create', 'delete', 'link', 'mkdir', 'modify', 'rmdir' ] )
  }
 # FIXME: doesn't work... wonder why?
@@ -1607,6 +1608,9 @@ def parse_line(self, component, cfg, cfname, lineno, l ):
                 setattr(self, k, isTrue(v))
             if k in ['logReject'] and self.logReject:
                 self.logEvents = self.logEvents | set(['reject'])
+
+            if k in ['logDuplicates'] and self.logDuplicates:
+                self.logEvents = self.logEvents | set(['nodupe'])
             return
 
         if len(line) < 2:

sarracenia/flow/__init__.py

@@ -1070,7 +1070,7 @@ def filter(self) -> None:
                                          (m['fileOp']['rename']))
                         else:
                             self.reject(
-                                m, 304, "mask=%s strip=%s url=%s" %
+                                m, 404, "mask=%s strip=%s url=%s" %
                                 (str(mask), strip, urlToMatch))
                         break
 
@@ -1105,7 +1105,7 @@ def filter(self) -> None:
                                            self.o.flatten)
                     filtered_worklist.append(m)
                 else:
-                    self.reject(m, 304, "unmatched pattern %s" % url)
+                    self.reject(m, 404, "unmatched pattern %s" % url)
 
         self.worklist.incoming = filtered_worklist
 
@@ -1479,7 +1479,6 @@ def file_should_be_downloaded(self, msg) -> bool:
                     logger.debug("%s file size different, so cannot be the same" %
                              (msg['new_path']))
                     return True
-
             else:
                 end = 0
 
@@ -1499,7 +1498,7 @@ def file_should_be_downloaded(self, msg) -> bool:
                         pass
 
                 if new_mtime <= old_mtime:
-                    self.reject(msg, 304,
+                    self.reject(msg, 406,
                             "mtime not newer %s " % (msg['new_path']))
                     return False
                 else:

sarracenia/flow/winnow.py

@@ -6,6 +6,7 @@
 default_options = {
         'acceptUnmatched': True, 
         'nodupe_ttl': 300,
+        'logDuplicates': True
 }
 
 

sarracenia/flowcb/log.py

@@ -159,6 +159,13 @@ def after_accept(self, worklist):
                         (msg['relPath'], msg['report']['code'], msg['report']['message']))
                 else:
                     logger.info("rejected: %s " % self._messageAcceptStr(msg))
+
+        elif 'nodupe' in self.o.logEvents:
+            for msg in worklist.rejected:
+                if 'report' in msg and msg['report']['code'] in [ 304 ]:
+                    logger.info(
+                        "%s rejected: %d %s " %
+                        (msg.getIDStr(), msg['report']['code'], msg['report']['message']))
 
         for msg in worklist.incoming:
 
@@ -188,6 +195,13 @@ def after_work(self, worklist):
                 else:
                     logger.info("rejected: %s " % self._messageStr(msg))
 
+        elif 'nodupe' in self.o.logEvents:
+            for msg in worklist.rejected:
+                if 'report' in msg and msg['report']['code'] in [ 304 ]:
+                    logger.info(
+                        "%s rejected: %d %s " %
+                        (msg.getIDStr(), msg['report']['code'], msg['report']['message']))
+
         for msg in worklist.ok:
             if 'size' in msg:
                 self.fileBytes += msg['size']

sarracenia/flowcb/nodupe/disk.py

@@ -163,22 +163,22 @@ def after_accept(self, worklist):
                 if mtime < min_mtime:
                     m['_deleteOnPost'] |= set(['reject'])
                     m['reject'] = f"{m['mtime']} too old (nodupe check), oldest allowed {timeflt2str(min_mtime)}"
-                    m.setReport(304,  f"{m['mtime']} too old (nodupe check), oldest allowed {timeflt2str(min_mtime)}" )
+                    m.setReport(406,  f"{m['mtime']} too old (nodupe check), oldest allowed {timeflt2str(min_mtime)}" )
                     worklist.rejected.append(m)
                     continue
                 elif mtime > max_mtime:
                     m['_deleteOnPost'] |= set(['reject'])
                     m['reject'] = f"{m['mtime']} too new (nodupe check), newest allowed {timeflt2str(max_mtime)}"
-                    m.setReport(304,  f"{m['mtime']} too new (nodupe check), newest allowed {timeflt2str(max_mtime)}" )
+                    m.setReport(425,  f"{m['mtime']} too new (nodupe check), newest allowed {timeflt2str(max_mtime)}" )
                     worklist.rejected.append(m)
                     continue
 
             if '_isRetry' in m or self.check_message(m):
                 new_incoming.append(m)
             else:
                 m['_deleteOnPost'] |= set(['reject'])
-                m['reject'] = "not modifified 1 (nodupe check)"
-                m.setReport(304, 'Not modified 1 (cache check)')
+                m['reject'] = "not modified 1 (nodupe check)"
+                m.setReport(304, 'Not modified 1 (nodupe check)')
                 worklist.rejected.append(m)
 
         if self.fp:

sarracenia/flowcb/nodupe/redis.py

@@ -163,22 +163,22 @@ def after_accept(self, worklist):
                 if mtime < min_mtime:
                     m['_deleteOnPost'] |= set(['reject'])
                     m['reject'] = f"{m['mtime']} too old (nodupe check), oldest allowed {timeflt2str(min_mtime)}"
-                    m.setReport(304,  f"{m['mtime']} too old (nodupe check), oldest allowed {timeflt2str(min_mtime)}" )
+                    m.setReport(406,  f"{m['mtime']} too old (nodupe check), oldest allowed {timeflt2str(min_mtime)}" )
                     worklist.rejected.append(m)
                     continue
                 elif mtime > max_mtime:
                     m['_deleteOnPost'] |= set(['reject'])
                     m['reject'] = f"{m['mtime']} too new (nodupe check), newest allowed {timeflt2str(max_mtime)}"
-                    m.setReport(304,  f"{m['mtime']} too new (nodupe check), newest allowed {timeflt2str(max_mtime)}" )
+                    m.setReport(425,  f"{m['mtime']} too new (nodupe check), newest allowed {timeflt2str(max_mtime)}" )
                     worklist.rejected.append(m)
                     continue
 
             if '_isRetry' in m or self._is_new(m):
                 new_incoming.append(m)
             else:
                 m['_deleteOnPost'] |= set(['reject'])
-                m['reject'] = "not modifified 1 (nodupe check)"
-                m.setReport(304, 'Not modified 1 (cache check)')
+                m['reject'] = "not modified 1 (nodupe check)"
+                m.setReport(304, 'Not modified 1 (nodupe check)')
                 worklist.rejected.append(m)
 
         logger.debug("items registered in duplicate suppression cache: %d" % (len(self._redis.keys(self._rkey_base + ":*"))) )

sarracenia/moth/__init__.py

@@ -284,6 +284,10 @@ def __init__(self, props=None, is_subscriber=True) -> None:
         self.metrics = { 'connected': False }
         self.metricsReset()
 
+        now = time.time()
+        self.next_connect_time = now
+        self.next_connect_failures = 0
+
         if (sys.version_info.major == 3) and (sys.version_info.minor < 7):
             self.o = {}
             for k in default_options:
@@ -415,6 +419,23 @@ def cleanup(self) -> None:
         else:
             self.putCleanUp()
 
+    def setEbo(self,start)->None:
+        """  Calculate next retry time using exponential backoff
+             note that it doesn't look like classic EBO because the time
+             is multiplied by how long it took to fail. Long failures should not
+             be retried quickly, but short failures can be variable in duration.
+             If the timing of failures is variable, the "attempt_duration" will be low,
+             and so the next_try might get smaller even though it hasn't succeeded yet...
+             it should eventually settle down to a long period though.
+        """
+        now=time.time()
+        attempt_duration = now - start
+        self.next_connect_failures += 1
+        ebo = 2**self.next_connect_failures
+        next_try = min(attempt_duration * ebo, 600)
+        self.next_connect_time = now + next_try
+        logger.error( f"could not connect. next try in {next_try} seconds.")
+
 if features['amqp']['present']:
     import sarracenia.moth.amqp
     import sarracenia.moth.amqpconsumer