add capability to strip outgoing changesets in opengrok-mirror

fixes #3892

Author: vladak

tools/src/main/python/opengrok_tools/mirror.py

@@ -19,7 +19,7 @@
 #
 
 #
-# Copyright (c) 2018, 2021, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2018, 2022, Oracle and/or its affiliates. All rights reserved.
 # Portions Copyright (c) 2019, Krystof Tulinger <[email protected]>
 #
 
@@ -58,25 +58,26 @@
 if major_version < 3:
     fatal("Need Python 3, you are running {}".format(major_version))
 
-__version__ = "1.1"
+__version__ = "1.2"
 
 OPENGROK_NO_MIRROR_ENV = "OPENGROK_NO_MIRROR"
 
 
 def worker(args):
-    project_name, logdir, loglevel, backupcount, config, check_changes, uri, \
-        source_root, batch, headers, api_timeout = args
+    project_name, logdir, loglevel, backup_count, config, check_changes, check_outgoing, uri, \
+        source_root, batch, headers, timeout, api_timeout = args
 
     if batch:
         get_batch_logger(logdir, project_name,
                          loglevel,
-                         backupcount,
+                         backup_count,
                          get_class_basename())
 
     return mirror_project(config, project_name,
-                          check_changes,
+                          check_changes, check_outgoing,
                           uri, source_root, headers=headers,
-                          timeout=api_timeout)
+                          timeout=timeout,
+                          api_timeout=api_timeout)
 
 
 def main():
@@ -107,12 +108,18 @@ def main():
                              ' repositories,'
                              ' terminate the processing'
                              ' if no change is found.')
+    parser.add_argument('--strip_outgoing', type=bool, default=False,
+                        help='check outgoing changes for each repository of a project,'
+                        'strip any such changes and remove all project data so that'
+                        'it can be reindexed from scratch')
     parser.add_argument('-w', '--workers', default=cpu_count(), type=int,
                         help='Number of worker processes')
     add_http_headers(parser)
     parser.add_argument('--api_timeout', type=int, default=3,
                         help='Set response timeout in seconds '
                              'for RESTful API calls')
+    parser.add_argument('--async_api_timeout', type=int, default=300,
+                        help='Set timeout in seconds for asynchronous RESTful API calls')
 
     try:
         args = parser.parse_args()
@@ -209,9 +216,11 @@ def main():
                     worker_args.append([x, logdir, args.loglevel,
                                         args.backupcount, config,
                                         args.check_changes,
+                                        args.strip_outgoing,
                                         args.uri, source_root,
                                         args.batch, headers,
-                                        args.api_timeout])
+                                        args.api_timeout,
+                                        args.async_api_timeout])
                 try:
                     project_results = pool.map(worker, worker_args, 1)
                 except KeyboardInterrupt:

tools/src/main/python/opengrok_tools/scm/git.py

@@ -18,7 +18,7 @@
 #
 
 #
-# Copyright (c) 2018, 2021, Oracle and/or its affiliates. All rights reserved.
+# Copyright (c) 2018, 2022, Oracle and/or its affiliates. All rights reserved.
 # Portions Copyright (c) 2020, Krystof Tulinger <[email protected]>
 #
 
@@ -54,3 +54,25 @@ def reposync(self):
     def incoming_check(self):
         self._configure_git_pull()
         return self._run_custom_incoming_command([self.command, 'pull', '--dry-run'])
+
+    def strip_outgoing(self):
+        self._configure_git_pull()
+        status, out = self._run_command([self.command, 'log',
+                                        '--pretty=tformat:%H', '--reverse', 'origin..'])
+        if status == 0:
+            cset = out.get(0)
+            if cset:
+                self.logger.debug("Resetting the repository {} to parent of changeset {}".
+                                  format(self, cset))
+                status, out = self._run_command([self.command, 'reset', '--hard',
+                                                 cset + '^'])
+                if status != 0:
+                    raise RepositoryException("failed to reset {} to parent of changeset {}: {}".
+                                              format(self, cset, out))
+                else:
+                    return True
+            else:
+                return False
+
+        raise RepositoryException("failed to check for outgoing changes in {}: {}".
+                                  format(self, status))

tools/src/main/python/opengrok_tools/scm/repository.py

@@ -105,13 +105,23 @@ def incoming(self):
     def incoming_check(self):
         """
         Check if there are any incoming changes.
-        Normally this method definition is overriden, unless the repository
+        Normally this method definition is overridden, unless the repository
         type has no way how to check for incoming changes.
 
-        Return True if so, False otherwise.
+        :return True if so, False otherwise.
         """
         return True
 
+    def strip_outgoing(self):
+        """
+        Strip any outgoing changes.
+        Normally this method definition is overridden, unless the repository
+        type has no way how to check for outgoing changes or cannot strip them.
+
+        :return True if any changes were stripped, False otherwise.
+        """
+        return False
+
     def _run_custom_sync_command(self, command):
         """
         Execute the custom sync command.
@@ -152,7 +162,7 @@ def _run_command(self, command):
                                env_vars=self.env, logger=self.logger)
         cmd.execute()
         if cmd.getretcode() != 0 or cmd.getstate() != Command.FINISHED:
-            cmd.log_error("failed to perform command")
+            cmd.log_error("failed to perform command {}".format(command))
             status = cmd.getretcode()
             if status == 0 and cmd.getstate() != Command.FINISHED:
                 status = 1

tools/src/main/python/opengrok_tools/utils/mirror.py

@@ -38,7 +38,7 @@
 )
 from .patterns import PROJECT_SUBST, COMMAND_PROPERTY
 from .utils import is_exe, check_create_dir, get_int, is_web_uri, get_bool
-from .opengrok import get_repos, get_repo_type, get_uri
+from .opengrok import get_repos, get_repo_type, get_uri, delete_project_data
 from .hook import run_hook
 from .command import Command
 from .restful import call_rest_api, do_api_call
@@ -64,6 +64,7 @@
 DISABLED_CMD_PROPERTY = 'disabled_command'
 HOOK_PRE_PROPERTY = "pre"
 HOOK_POST_PROPERTY = "post"
+STRIP_OUTGOING_PROPERTY = "strip_outgoing"
 
 
 def get_repos_for_project(project_name, uri, source_root,
@@ -177,6 +178,7 @@ def get_project_properties(project_config, project_name, hookdir):
     use_proxy = False
     ignored_repos = None
     check_changes = None
+    check_outgoing = None
     ignore_errors = None
 
     logger = logging.getLogger(__name__)
@@ -230,6 +232,12 @@ def get_project_properties(project_config, project_name, hookdir):
                                      project_config.get(INCOMING_PROPERTY))
             logger.debug("incoming check = {}".format(check_changes))
 
+        if project_config.get(STRIP_OUTGOING_PROPERTY) is not None:
+            check_outgoing = get_bool(logger, ("outgoing check for project {}".
+                                               format(project_name)),
+                                      project_config.get(STRIP_OUTGOING_PROPERTY))
+            logger.debug("outgoing check = {}".format(check_changes))
+
         if project_config.get(IGNORE_ERR_PROPERTY) is not None:
             ignore_errors = get_bool(logger, ("ignore errors for project {}".
                                               format(project_name)),
@@ -240,7 +248,7 @@ def get_project_properties(project_config, project_name, hookdir):
         ignored_repos = []
 
     return prehook, posthook, hook_timeout, command_timeout, \
-        use_proxy, ignored_repos, check_changes, ignore_errors
+        use_proxy, ignored_repos, check_changes, check_outgoing, ignore_errors
 
 
 def process_hook(hook_ident, hook, source_root, project_name, proxy,
@@ -342,7 +350,7 @@ def run_command(cmd, project_name):
 
 
 def handle_disabled_project(config, project_name, disabled_msg, headers=None,
-                            timeout=None):
+                            timeout=None, api_timeout=None):
     disabled_command = config.get(DISABLED_CMD_PROPERTY)
     if disabled_command:
         logger = logging.getLogger(__name__)
@@ -365,7 +373,7 @@ def handle_disabled_project(config, project_name, disabled_msg, headers=None,
 
             try:
                 call_rest_api(disabled_command, {PROJECT_SUBST: project_name},
-                              http_headers=headers, timeout=timeout)
+                              http_headers=headers, timeout=timeout, api_timeout=api_timeout)
             except RequestException as e:
                 logger.error("API call failed for disabled command of "
                              "project '{}': {}".
@@ -393,18 +401,58 @@ def get_mirror_retcode(ignore_errors, value):
     return value
 
 
-def mirror_project(config, project_name, check_changes, uri,
-                   source_root, headers=None, timeout=None):
+def process_outgoing(repos, project_name):
+    """
+    Detect and strip any outgoing changes for the repositories.
+    :param repos: list of repository objects
+    :param project_name: name of the project
+    :return: if any of the repositories had to be reset
+    """
+
+    logger = logging.getLogger(__name__)
+
+    ret = False
+    for repo in repos:
+        if repo.strip_outgoing():
+            logger.debug('Repository {} in project {} had outgoing changes stripped'.
+                         format(repo, project_name))
+            ret = True
+
+    return ret
+
+
+def wipe_project_data(project_name, uri, headers=None, timeout=None, api_timeout=None):
+    """
+    Remove data for the project and mark it as not indexed.
+    :param project_name: name of the project
+    :param uri: URI of the webapp
+    :param headers: HTTP headers
+    :param timeout: connect timeout
+    :param api_timeout: asynchronous API timeout
+    """
+
+    logger = logging.getLogger(__name__)
+
+    logger.info("removing data for project {}".format(project_name))
+    delete_project_data(logger, project_name, uri,
+                        headers=headers, timeout=timeout, api_timeout=api_timeout)
+
+
+def mirror_project(config, project_name, check_changes, check_outgoing, uri,
+                   source_root, headers=None, timeout=None, api_timeout=None):
     """
     Mirror the repositories of single project.
     :param config global configuration dictionary
     :param project_name: name of the project
     :param check_changes: check for changes in the project or its repositories
      and terminate if no change is found
+    :param check_outgoing: check for outgoing changes in the repositories of the project,
+     strip the changes and wipe project data if such changes were found
     :param uri web application URI
     :param source_root source root
     :param headers: optional dictionary of HTTP headers
-    :param timeout: optional timeout in seconds for API call response
+    :param timeout: connect timeout
+    :param api_timeout: optional timeout in seconds for API call response
     :return exit code
     """
 
@@ -416,6 +464,7 @@ def mirror_project(config, project_name, check_changes, uri,
     prehook, posthook, hook_timeout, command_timeout, use_proxy, \
         ignored_repos, \
         check_changes_proj, \
+        check_outgoing_proj, \
         ignore_errors_proj = get_project_properties(project_config,
                                                     project_name,
                                                     config.
@@ -431,6 +480,11 @@ def mirror_project(config, project_name, check_changes, uri,
     else:
         check_changes_config = check_changes_proj
 
+    if check_outgoing_proj is None:
+        check_outgoing_config = config.get(STRIP_OUTGOING_PROPERTY)
+    else:
+        check_outgoing_config = check_outgoing_proj
+
     if ignore_errors_proj is None:
         ignore_errors = config.get(IGNORE_ERR_PROPERTY)
     else:
@@ -446,7 +500,8 @@ def mirror_project(config, project_name, check_changes, uri,
                                 project_config.
                                 get(DISABLED_REASON_PROPERTY),
                                 headers=headers,
-                                timeout=timeout)
+                                timeout=timeout,
+                                api_timeout=api_timeout)
         logger.info("Project '{}' disabled, exiting".
                     format(project_name))
         return CONTINUE_EXITVAL
@@ -473,6 +528,20 @@ def mirror_project(config, project_name, check_changes, uri,
     if check_changes_config is not None:
         check_changes = check_changes_config
 
+    if check_outgoing_config is not None:
+        check_outgoing = check_outgoing_config
+
+    if check_outgoing:
+        try:
+            r = process_outgoing(repos, project_name)
+        except RepositoryException as exc:
+            logger.error('Failed to handle outgoing changes for '
+                         'a repository in project {}: {}'.format(project_name, exc))
+            return get_mirror_retcode(ignore_errors, FAILURE_EXITVAL)
+        if r:
+            wipe_project_data(project_name, uri, headers=headers,
+                              timeout=timeout, api_timeout=api_timeout)
+
     # Check if the project or any of its repositories have changed.
     if check_changes:
         r = process_changes(repos, project_name, uri, headers=headers)
@@ -522,7 +591,7 @@ def check_project_configuration(multiple_project_config, hookdir=False,
                               HOOK_TIMEOUT_PROPERTY, PROXY_PROPERTY,
                               IGNORED_REPOS_PROPERTY, HOOKS_PROPERTY,
                               DISABLED_REASON_PROPERTY, INCOMING_PROPERTY,
-                              IGNORE_ERR_PROPERTY]
+                              IGNORE_ERR_PROPERTY, STRIP_OUTGOING_PROPERTY]
 
     if not multiple_project_config:
         return True
@@ -640,7 +709,7 @@ def check_configuration(config):
                        COMMANDS_PROPERTY, PROJECTS_PROPERTY,
                        HOOK_TIMEOUT_PROPERTY, CMD_TIMEOUT_PROPERTY,
                        DISABLED_CMD_PROPERTY, INCOMING_PROPERTY,
-                       IGNORE_ERR_PROPERTY]
+                       IGNORE_ERR_PROPERTY, STRIP_OUTGOING_PROPERTY]
 
     diff = set(config.keys()).difference(global_tunables)
     if diff:

tools/src/main/python/opengrok_tools/utils/opengrok.py

@@ -190,10 +190,9 @@ def add_project(logger, project, uri, headers=None, timeout=None, api_timeout=No
     return True
 
 
-def delete_project(logger, project, uri, headers=None, timeout=None, api_timeout=None):
+def _delete_project(logger, project, uri, headers=None, timeout=None, api_timeout=None):
     try:
-        r = do_api_call('DELETE', get_uri(uri, 'api', 'v1', 'projects',
-                                          urllib.parse.quote_plus(project)),
+        r = do_api_call('DELETE', uri,
                         headers=headers, timeout=timeout, api_timeout=api_timeout)
         if r is None or r.status_code != 204:
             logger.error(f"could not delete project '{project}' in web application")
@@ -204,3 +203,17 @@ def delete_project(logger, project, uri, headers=None, timeout=None, api_timeout
         return False
 
     return True
+
+
+def delete_project(logger, project, uri, headers=None, timeout=None, api_timeout=None):
+    return _delete_project(logger, project, get_uri(uri, 'api', 'v1', 'projects',
+                                                    urllib.parse.quote_plus(project)),
+                           headers=headers,
+                           timeout=timeout, api_timeout=api_timeout)
+
+
+def delete_project_data(logger, project, uri, headers=None, timeout=None, api_timeout=None):
+    return _delete_project(logger, project, get_uri(uri, 'api', 'v1', 'projects',
+                                                    urllib.parse.quote_plus(project), 'data'),
+                           headers=headers,
+                           timeout=timeout, api_timeout=api_timeout)