(self, args: List[str],
stdout_as_json: bool = True)
| 2761 | |
| 2762 | @API.expose |
| 2763 | def send_rgwadmin_command(self, args: List[str], |
| 2764 | stdout_as_json: bool = True) -> Tuple[int, Union[str, dict], str]: |
| 2765 | try: |
| 2766 | cmd = [ |
| 2767 | 'radosgw-admin', |
| 2768 | '-c', str(self.get_ceph_conf_path()), |
| 2769 | '-k', str(self.get_ceph_option('keyring')), |
| 2770 | '-n', f'mgr.{self.get_mgr_id()}', |
| 2771 | ] + args |
| 2772 | self.log.debug('Executing %s', str(cmd)) |
| 2773 | result = subprocess.run( # pylint: disable=subprocess-run-check |
| 2774 | cmd, |
| 2775 | stdout=subprocess.PIPE, |
| 2776 | stderr=subprocess.PIPE, |
| 2777 | timeout=10, |
| 2778 | ) |
| 2779 | stdout = result.stdout.decode('utf-8') |
| 2780 | stderr = result.stderr.decode('utf-8') |
| 2781 | if stdout and stdout_as_json: |
| 2782 | stdout = json.loads(stdout) |
| 2783 | if result.returncode: |
| 2784 | self.log.debug('Error %s executing %s: %s', result.returncode, str(cmd), stderr) |
| 2785 | return result.returncode, stdout, stderr |
| 2786 | except subprocess.CalledProcessError as ex: |
| 2787 | self.log.exception('Error executing radosgw-admin %s: %s', str(ex.cmd), str(ex.output)) |
| 2788 | raise |
| 2789 | except subprocess.TimeoutExpired as ex: |
| 2790 | self.log.error('Timeout (10s) executing radosgw-admin %s', str(ex.cmd)) |
| 2791 | raise |
no test coverage detected