(self, api, base_tags)
A registry here is an external docker registry (DockerHub, ECR, another Harbor...) that this current
        instance of Harbor is using for replication purposes. The health of those services can be monitored within
        Harbor API.
A registry here is an external docker registry (DockerHub, ECR, another Harbor...) that this current
        instance of Harbor is using for replication purposes. The health of those services can be monitored within
        Harbor API.
A registry here is an external docker registry ( DockerHub ECR another Harbor ... ) that this current instance of Harbor is using for replication purposes . The health of those services can be monitored within Harbor API .
def _check_registries_health(self, api, base_tags):
        """A registry here is an external docker registry (DockerHub, ECR, another Harbor...) that this current
        instance of Harbor is using for replication purposes. The health of those services can be monitored within
        Harbor API."""
        try:
            registries = api.registries()
            self.log.debug("Found %d registries", len(registries))
        except HTTPError as e:
            if e.response.status_code in (401, 403):
                # Forbidden, user is not admin
                self.log.info(
                    "Provided user in harbor integration config is not an admin user. Ignoring registries health checks"
                )
                self.log.debug(e, exc_info=True)
                return
            raise e

        for registry in registries:
            registry_name = registry['name'].lower()
            tags = base_tags + ['registry:' + registry_name]
            if registry.get('status'):
                status = AgentCheck.OK if registry['status'] == HEALTHY else AgentCheck.CRITICAL
                self.service_check(REGISTRY_STATUS, status, tags=tags)
            else:
                try:
                    api.registry_health(registry['id'])
                    self.service_check(REGISTRY_STATUS, AgentCheck.OK, tags=tags)
                except HTTPError as e:
                    self.log.debug(e, exc_info=True)
                    self.service_check(REGISTRY_STATUS, AgentCheck.CRITICAL, tags=tags)
def _check_registries_health ( self , api , base_tags ) : try : registries = api . registries ( ) self . log . debug ( "Found %d registries" , len ( registries ) ) except HTTPError as e : if e . response . status_code in ( 401 , 403 ) : # Forbidden, user is not admin self . log . info ( "Provided user in harbor integration config is not an admin user. Ignoring registries health checks" ) self . log . debug ( e , exc_info = True ) return raise e for registry in registries : registry_name = registry [ 'name' ] . lower ( ) tags = base_tags + [ 'registry:' + registry_name ] if registry . get ( 'status' ) : status = AgentCheck . OK if registry [ 'status' ] == HEALTHY else AgentCheck . CRITICAL self . service_check ( REGISTRY_STATUS , status , tags = tags ) else : try : api . registry_health ( registry [ 'id' ] ) self . service_check ( REGISTRY_STATUS , AgentCheck . OK , tags = tags ) except HTTPError as e : self . log . debug ( e , exc_info = True ) self . service_check ( REGISTRY_STATUS , AgentCheck . CRITICAL , tags = tags )