ops: make ping view return orderedDict, and make the platform-id expose variables

rhodecode/apps/ops/views.py

@@ -1,97 +1,96 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import time
 import logging
 
 
 from pyramid.httpexceptions import HTTPFound
 
 from rhodecode.apps._base import BaseAppView
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import LoginRequired
+from rhodecode.lib.compat import OrderedDict
 from rhodecode.model.db import UserApiKeys
 
 log = logging.getLogger(__name__)
 
 
 class OpsView(BaseAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         c.user = c.auth_user.get_instance()
 
         return c
 
     def ops_ping(self):
-        data = {
-            'instance': self.request.registry.settings.get('instance_id'),
-        }
+        data = OrderedDict()
+        data['instance'] = self.request.registry.settings.get('instance_id')
+
         if getattr(self.request, 'user'):
             caller_name = 'anonymous'
             if self.request.user.user_id:
                 caller_name = self.request.user.username
 
-            data.update({
-                'caller_ip': self.request.user.ip_addr,
-                'caller_name': caller_name,
-            })
+            data['caller_ip'] = self.request.user.ip_addr
+            data['caller_name'] = caller_name
+
         return {'ok': data}
 
     def ops_error_test(self):
         """
         Test exception handling and emails on errors
         """
 
         class TestException(Exception):
             pass
         # add timeout so we add some sort of rate limiter
         time.sleep(2)
         msg = ('RhodeCode Enterprise test exception. '
                'Client:{}. Generation time: {}.'.format(self.request.user, time.time()))
         raise TestException(msg)
 
     def ops_redirect_test(self):
         """
         Test redirect handling
         """
         redirect_to = self.request.GET.get('to') or h.route_path('home')
         raise HTTPFound(redirect_to)
 
     @LoginRequired(auth_token_access=[UserApiKeys.ROLE_HTTP])
     def ops_healthcheck(self):
         from rhodecode.lib.system_info import load_system_info
 
         vcsserver_info = load_system_info('vcs_server')
         if vcsserver_info:
             vcsserver_info = vcsserver_info['human_value']
 
         db_info = load_system_info('database_info')
         if db_info:
             db_info = db_info['human_value']
 
         health_spec = {
             'caller_ip': self.request.user.ip_addr,
             'vcsserver': vcsserver_info,
             'db': db_info,
         }
 
         return {'healthcheck': health_spec}
-

rhodecode/config/utils.py

@@ -1,101 +1,104 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 import platform
 
 from rhodecode.model import init_model
 
 
 def configure_vcs(config):
     """
     Patch VCS config with some RhodeCode specific stuff
     """
     from rhodecode.lib.vcs import conf
     import rhodecode.lib.vcs.conf.settings
 
     conf.settings.BACKENDS = {
         'hg': 'rhodecode.lib.vcs.backends.hg.MercurialRepository',
         'git': 'rhodecode.lib.vcs.backends.git.GitRepository',
         'svn': 'rhodecode.lib.vcs.backends.svn.SubversionRepository',
     }
 
     conf.settings.HOOKS_PROTOCOL = config['vcs.hooks.protocol']
     conf.settings.HOOKS_HOST = config['vcs.hooks.host']
     conf.settings.HOOKS_DIRECT_CALLS = config['vcs.hooks.direct_calls']
     conf.settings.DEFAULT_ENCODINGS = config['default_encoding']
     conf.settings.ALIASES[:] = config['vcs.backends']
     conf.settings.SVN_COMPATIBLE_VERSION = config['vcs.svn.compatible_version']
 
 
 def initialize_database(config):
     from rhodecode.lib.utils2 import engine_from_config, get_encryption_key
     engine = engine_from_config(config, 'sqlalchemy.db1.')
     init_model(engine, encryption_key=get_encryption_key(config))
 
 
 def initialize_test_environment(settings, test_env=None):
     if test_env is None:
         test_env = not int(os.environ.get('RC_NO_TMP_PATH', 0))
 
     from rhodecode.lib.utils import (
         create_test_directory, create_test_database, create_test_repositories,
         create_test_index)
     from rhodecode.tests import TESTS_TMP_PATH
     from rhodecode.lib.vcs.backends.hg import largefiles_store
     from rhodecode.lib.vcs.backends.git import lfs_store
 
     # test repos
     if test_env:
         create_test_directory(TESTS_TMP_PATH)
         # large object stores
         create_test_directory(largefiles_store(TESTS_TMP_PATH))
         create_test_directory(lfs_store(TESTS_TMP_PATH))
 
         create_test_database(TESTS_TMP_PATH, settings)
         create_test_repositories(TESTS_TMP_PATH, settings)
         create_test_index(TESTS_TMP_PATH, settings)
 
 
 def get_vcs_server_protocol(config):
     return config['vcs.server.protocol']
 
 
 def set_instance_id(config):
     """
     Sets a dynamic generated config['instance_id'] if missing or '*'
     E.g instance_id = *cluster-1 or instance_id = *
     """
 
     config['instance_id'] = config.get('instance_id') or ''
     instance_id = config['instance_id']
     if instance_id.startswith('*') or not instance_id:
         prefix = instance_id.lstrip('*')
         _platform_id = platform.uname()[1] or 'instance'
-        config['instance_id'] = '%s%s-%s' % (prefix, _platform_id, os.getpid())
+        config['instance_id'] = '{prefix}uname:{platform}-pid:{pid}'.format(
+            prefix=prefix,
+            platform=_platform_id,
+            pid=os.getpid())
 
 
 def get_default_user_id():
     from rhodecode.model.db import User, Session
     user_id = Session()\
         .query(User.user_id)\
         .filter(User.username == User.DEFAULT_USER)\
         .scalar()
     return user_id