plugin + housekeeping

[danilodorgam]

Do you know if the use of this plug-in interferes with cleaning the database via housekeeping?
Even after changing to one day the housekeeping has not taken effect.

Sorry if it has nothing to do with the plugin

[javamachr]

Do you mean DB of Sonar or DB of Gitlab?
AFAIK it should not have any impact on this as it only reports results to GitLab via API.

[danilodorgam]

it refers to the sonar db, we are using 100 gb in development and we wanted to reduce this value.

[javamachr]

I believe that this plugin in particular does not touch sonar DB. But it requires Community Branch plugin for proper work adn that one definitely does write into sonar DB(branches + PRs) so you may need to check that one.

[emericv]

I make my own housekeeper, you can put it in a scheduled GitLab job

Known limits:

• API result is limited to 100 elements by page: this tool should loop over all pages.

Code:

#! /usr/bin/env python
import argparse
import json
import os
import requests
import sys
import time
from dateutil import parser as date_parser
from dateutil.relativedelta import relativedelta
from datetime import datetime, timezone
from requests.auth import HTTPBasicAuth

if 'SONAR_HOST_URL' in os.environ:
    SONAR_HOST_URL = str(os.environ.get('SONAR_HOST_URL'))
else:
    SONAR_HOST_URL = 'https://sonar.example.com'

if 'SONAR_TOKEN' in os.environ:
    SONAR_TOKEN = str(os.environ.get('SONAR_TOKEN'))
else:
    SONAR_TOKEN = ''

if 'DRY_RUN' in os.environ:
    DRY_RUN = bool(os.environ.get('DRY_RUN'))
else:
    DRY_RUN = False

if 'BRANCH_AGE_LIMIT_IN_DAYS' in os.environ:
    BRANCH_AGE_LIMIT_IN_DAYS = int(os.environ.get('BRANCH_AGE_LIMIT_IN_DAYS'))
else:
    BRANCH_AGE_LIMIT_IN_DAYS = 30

class Error(Exception):
    """Error message"""

class SonarqubeHousekeeper:
    # SonarQube Housekeeper
    
    def __init__(self, sonar_host_url=SONAR_HOST_URL, sonar_token=SONAR_TOKEN, dry_run=False):
        # Initialize the housekeeper
        self.sonar_api_url = sonar_host_url + '/api'
        self.sonar_token = sonar_token
        self.dry_run = dry_run
        self.headers = {}

    def _request(self, path, method, params = {}):
        # Process API request
        response = requests.request(method, headers=self.headers,
            url=self.sonar_api_url + path, params=params, auth=HTTPBasicAuth(self.sonar_token, '')
        )
        response.raise_for_status()
        if response.headers.get('Content-Type') is not None:
            if response.headers.get('content-type') == 'application/json':
                return response.json()

    def get_project_list(self):
        # Get the project list
        data = self._request(path='/projects/search', method='GET', params={'ps': '100'})
        iterator = map(lambda item: item['key'], data['components'])
        return list(iterator)

    def get_project_stale_branch_list(self, project_id):
        # Get the project list
        data = self._request(path='/project_branches/list', method='GET',
                             params={'ps': '100', 'project': project_id})
        stale_branch_list = []
        date_limit = datetime.now(timezone.utc) - relativedelta(days=BRANCH_AGE_LIMIT_IN_DAYS)
        for branch in data['branches']:
            if branch['excludedFromPurge'] or branch['type'] != 'BRANCH':
                continue
            analysis_time = date_parser.parse(branch['analysisDate'])
            if analysis_time > date_limit:
                continue
            stale_branch_list.append(branch['name'])
        return stale_branch_list

    def remove_project_branch(self, project_id, branch_name):
        # Remove branch
        print('Removing stale branch [' + project_id + ' - ' + branch_name + ']...')
        if self.dry_run:
            return
        self._request(path='/project_branches/delete', method='POST',
                      params={'project': project_id, 'branch': branch_name})

    def remove_project_stale_branches(self, project_id):
        # Remove stale branches
        branch_list = self.get_project_stale_branch_list(project_id)
        for branch in branch_list:
            self.remove_project_branch(project_id, branch)

    def remove_all_stale_branches(self):
        # Remove stale branches
        project_list = self.get_project_list()
        for project_id in project_list:
            self.remove_project_stale_branches(project_id)


if __name__ == "__main__":
    housekeeper = SonarqubeHousekeeper(sonar_token=SONAR_TOKEN, dry_run=DRY_RUN)
    housekeeper.remove_all_stale_branches()