"""The ``celery control``, ``. inspect`` and ``. status`` programs."""
from functools import partial

import click
from kombu.utils.json import dumps

from celery.bin.base import (COMMA_SEPARATED_LIST, CeleryCommand,
                             CeleryOption, handle_preload_options)
from celery.exceptions import CeleryCommandException
from celery.platforms import EX_UNAVAILABLE
from celery.utils import text
from celery.worker.control import Panel


def _say_remote_command_reply(ctx, replies, show_reply=False):
    node = next(iter(replies))  # <-- take first.
    reply = replies[node]
    node = ctx.obj.style(f'{node}: ', fg='cyan', bold=True)
    status, preply = ctx.obj.pretty(reply)
    ctx.obj.say_chat('->', f'{node}{status}',
                     text.indent(preply, 4) if show_reply else '',
                     show_body=show_reply)


def _consume_arguments(meta, method, args):
    i = 0
    try:
        for i, arg in enumerate(args):
            try:
                name, typ = meta.args[i]
            except IndexError:
                if meta.variadic:
                    break
                raise click.UsageError(
                    'Command {!r} takes arguments: {}'.format(
                        method, meta.signature))
            else:
                yield name, typ(arg) if typ is not None else arg
    finally:
        args[:] = args[i:]


def _compile_arguments(action, args):
    meta = Panel.meta[action]
    arguments = {}
    if meta.args:
        arguments.update({
            k: v for k, v in _consume_arguments(meta, action, args)
        })
    if meta.variadic:
        arguments.update({meta.variadic: args})
    return arguments


@click.command(cls=CeleryCommand)
@click.option('-t',
              '--timeout',
              cls=CeleryOption,
              type=float,
              default=1.0,
              help_group='Remote Control Options',
              help='Timeout in seconds waiting for reply.')
@click.option('-d',
              '--destination',
              cls=CeleryOption,
              type=COMMA_SEPARATED_LIST,
              help_group='Remote Control Options',
              help='Comma separated list of destination node names.')
@click.option('-j',
              '--json',
              cls=CeleryOption,
              is_flag=True,
              help_group='Remote Control Options',
              help='Use json as output format.')
@click.pass_context
@handle_preload_options
def status(ctx, timeout, destination, json, **kwargs):
    """Show list of workers that are online."""
    callback = None if json else partial(_say_remote_command_reply, ctx)
    replies = ctx.obj.app.control.inspect(timeout=timeout,
                                          destination=destination,
                                          callback=callback).ping()

    if not replies:
        raise CeleryCommandException(
            message='No nodes replied within time constraint',
            exit_code=EX_UNAVAILABLE
        )

    if json:
        ctx.obj.echo(dumps(replies))
    nodecount = len(replies)
    if not kwargs.get('quiet', False):
        ctx.obj.echo('\n{} {} online.'.format(
            nodecount, text.pluralize(nodecount, 'node')))


@click.command(cls=CeleryCommand,
               context_settings={'allow_extra_args': True})
@click.argument("action", type=click.Choice([
    name for name, info in Panel.meta.items()
    if info.type == 'inspect' and info.visible
]))
@click.option('-t',
              '--timeout',
              cls=CeleryOption,
              type=float,
              default=1.0,
              help_group='Remote Control Options',
              help='Timeout in seconds waiting for reply.')
@click.option('-d',
              '--destination',
              cls=CeleryOption,
              type=COMMA_SEPARATED_LIST,
              help_group='Remote Control Options',
              help='Comma separated list of destination node names.')
@click.option('-j',
              '--json',
              cls=CeleryOption,
              is_flag=True,
              help_group='Remote Control Options',
              help='Use json as output format.')
@click.pass_context
@handle_preload_options
def inspect(ctx, action, timeout, destination, json, **kwargs):
    """Inspect the worker at runtime.

    Availability: RabbitMQ (AMQP) and Redis transports.
    """
    callback = None if json else partial(_say_remote_command_reply, ctx,
                                         show_reply=True)
    arguments = _compile_arguments(action, ctx.args)
    inspect = ctx.obj.app.control.inspect(timeout=timeout,
                                          destination=destination,
                                          callback=callback)
    replies = inspect._request(action,
                               **arguments)

    if not replies:
        raise CeleryCommandException(
            message='No nodes replied within time constraint',
            exit_code=EX_UNAVAILABLE
        )

    if json:
        ctx.obj.echo(dumps(replies))
    nodecount = len(replies)
    if not ctx.obj.quiet:
        ctx.obj.echo('\n{} {} online.'.format(
            nodecount, text.pluralize(nodecount, 'node')))


@click.command(cls=CeleryCommand,
               context_settings={'allow_extra_args': True})
@click.argument("action", type=click.Choice([
    name for name, info in Panel.meta.items()
    if info.type == 'control' and info.visible
]))
@click.option('-t',
              '--timeout',
              cls=CeleryOption,
              type=float,
              default=1.0,
              help_group='Remote Control Options',
              help='Timeout in seconds waiting for reply.')
@click.option('-d',
              '--destination',
              cls=CeleryOption,
              type=COMMA_SEPARATED_LIST,
              help_group='Remote Control Options',
              help='Comma separated list of destination node names.')
@click.option('-j',
              '--json',
              cls=CeleryOption,
              is_flag=True,
              help_group='Remote Control Options',
              help='Use json as output format.')
@click.pass_context
@handle_preload_options
def control(ctx, action, timeout, destination, json):
    """Workers remote control.

    Availability: RabbitMQ (AMQP), Redis, and MongoDB transports.
    """
    callback = None if json else partial(_say_remote_command_reply, ctx,
                                         show_reply=True)
    args = ctx.args
    arguments = _compile_arguments(action, args)
    replies = ctx.obj.app.control.broadcast(action, timeout=timeout,
                                            destination=destination,
                                            callback=callback,
                                            reply=True,
                                            arguments=arguments)

    if not replies:
        raise CeleryCommandException(
            message='No nodes replied within time constraint',
            exit_code=EX_UNAVAILABLE
        )

    if json:
        ctx.obj.echo(dumps(replies))

celery/bin/control.py
"""The ``celery control``, ``. inspect`` and ``. status`` programs."""
from functools import partial

import click
from kombu.utils.json import dumps

from celery.bin.base import (COMMA_SEPARATED_LIST, CeleryCommand,
                             CeleryOption, handle_preload_options)
from celery.exceptions import CeleryCommandException
from celery.platforms import EX_UNAVAILABLE
from celery.utils import text
from celery.worker.control import Panel


def _say_remote_command_reply(ctx, replies, show_reply=False):
    node = next(iter(replies))  # <-- take first.
    reply = replies[node]
    node = ctx.obj.style(f'{node}: ', fg='cyan', bold=True)
    status, preply = ctx.obj.pretty(reply)
    ctx.obj.say_chat('->', f'{node}{status}',
                     text.indent(preply, 4) if show_reply else '',
                     show_body=show_reply)


def _consume_arguments(meta, method, args):
    i = 0
    try:
        for i, arg in enumerate(args):
            try:
                name, typ = meta.args[i]
            except IndexError:
                if meta.variadic:
                    break
                raise click.UsageError(
                    'Command {!r} takes arguments: {}'.format(
                        method, meta.signature))
            else:
                yield name, typ(arg) if typ is not None else arg
    finally:
        args[:] = args[i:]


def _compile_arguments(action, args):
    meta = Panel.meta[action]
    arguments = {}
    if meta.args:
        arguments.update({
            k: v for k, v in _consume_arguments(meta, action, args)
        })
    if meta.variadic:
        arguments.update({meta.variadic: args})
    return arguments


@click.command(cls=CeleryCommand)
@click.option('-t',
              '--timeout',
              cls=CeleryOption,
              type=float,
              default=1.0,
              help_group='Remote Control Options',
              help='Timeout in seconds waiting for reply.')
@click.option('-d',
              '--destination',
              cls=CeleryOption,
              type=COMMA_SEPARATED_LIST,
              help_group='Remote Control Options',
              help='Comma separated list of destination node names.')
@click.option('-j',
              '--json',
              cls=CeleryOption,
              is_flag=True,
              help_group='Remote Control Options',
              help='Use json as output format.')
@click.pass_context
@handle_preload_options
def status(ctx, timeout, destination, json, **kwargs):
    """Show list of workers that are online."""
    callback = None if json else partial(_say_remote_command_reply, ctx)
    replies = ctx.obj.app.control.inspect(timeout=timeout,
                                          destination=destination,
                                          callback=callback).ping()

    if not replies:
        raise CeleryCommandException(
            message='No nodes replied within time constraint',
            exit_code=EX_UNAVAILABLE
        )

    if json:
        ctx.obj.echo(dumps(replies))
    nodecount = len(replies)
    if not kwargs.get('quiet', False):
        ctx.obj.echo('\n{} {} online.'.format(
            nodecount, text.pluralize(nodecount, 'node')))


@click.command(cls=CeleryCommand,
               context_settings={'allow_extra_args': True})
@click.argument("action", type=click.Choice([
    name for name, info in Panel.meta.items()
    if info.type == 'inspect' and info.visible
]))
@click.option('-t',
              '--timeout',
              cls=CeleryOption,
              type=float,
              default=1.0,
              help_group='Remote Control Options',
              help='Timeout in seconds waiting for reply.')
@click.option('-d',
              '--destination',
              cls=CeleryOption,
              type=COMMA_SEPARATED_LIST,
              help_group='Remote Control Options',
              help='Comma separated list of destination node names.')
@click.option('-j',
              '--json',
              cls=CeleryOption,
              is_flag=True,
              help_group='Remote Control Options',
              help='Use json as output format.')
@click.pass_context
@handle_preload_options
def inspect(ctx, action, timeout, destination, json, **kwargs):
    """Inspect the worker at runtime.

    Availability: RabbitMQ (AMQP) and Redis transports.
    """
    callback = None if json else partial(_say_remote_command_reply, ctx,
                                         show_reply=True)
    arguments = _compile_arguments(action, ctx.args)
    inspect = ctx.obj.app.control.inspect(timeout=timeout,
                                          destination=destination,
                                          callback=callback)
    replies = inspect._request(action,
                               **arguments)

    if not replies:
        raise CeleryCommandException(
            message='No nodes replied within time constraint',
            exit_code=EX_UNAVAILABLE
        )

    if json:
        ctx.obj.echo(dumps(replies))
        return

    nodecount = len(replies)
    if not ctx.obj.quiet:
        ctx.obj.echo('\n{} {} online.'.format(
            nodecount, text.pluralize(nodecount, 'node')))


@click.command(cls=CeleryCommand,
               context_settings={'allow_extra_args': True})
@click.argument("action", type=click.Choice([
    name for name, info in Panel.meta.items()
    if info.type == 'control' and info.visible
]))
@click.option('-t',
              '--timeout',
              cls=CeleryOption,
              type=float,
              default=1.0,
              help_group='Remote Control Options',
              help='Timeout in seconds waiting for reply.')
@click.option('-d',
              '--destination',
              cls=CeleryOption,
              type=COMMA_SEPARATED_LIST,
              help_group='Remote Control Options',
              help='Comma separated list of destination node names.')
@click.option('-j',
              '--json',
              cls=CeleryOption,
              is_flag=True,
              help_group='Remote Control Options',
              help='Use json as output format.')
@click.pass_context
@handle_preload_options
def control(ctx, action, timeout, destination, json):
    """Workers remote control.

    Availability: RabbitMQ (AMQP), Redis, and MongoDB transports.
    """
    callback = None if json else partial(_say_remote_command_reply, ctx,
                                         show_reply=True)
    args = ctx.args
    arguments = _compile_arguments(action, args)
    replies = ctx.obj.app.control.broadcast(action, timeout=timeout,
                                            destination=destination,
                                            callback=callback,
                                            reply=True,
                                            arguments=arguments)

    if not replies:
        raise CeleryCommandException(
            message='No nodes replied within time constraint',
            exit_code=EX_UNAVAILABLE
        )

    if json:
        ctx.obj.echo(dumps(replies))

celery/bin/control.py
diff --git a/celery/bin/control.py b/celery/bin/control.py
index a13963a54b3..fbd3730c490 100644
--- a/celery/bin/control.py
+++ b/celery/bin/control.py
@@ -144,6 +144,8 @@ def inspect(ctx, action, timeout, destination, json, **kwargs):
 
     if json:
         ctx.obj.echo(dumps(replies))
+        return
+
     nodecount = len(replies)
     if not ctx.obj.quiet:
         ctx.obj.echo('\n{} {} online.'.format(

Celery inspect output JSON serialization is broken
# Checklist
- [ ] I have verified that the issue exists against the `master` branch of Celery.
- [ ] This has already been asked to the [discussion group](https://groups.google.com/forum/#!forum/celery-users) first.
- [x] I have read the relevant section in the
  [contribution guide](http://docs.celeryproject.org/en/latest/contributing.html#other-bugs)
  on reporting bugs.
- [x] I have checked the [issues list](https://github.com/celery/celery/issues?q=is%3Aissue+label%3A%22Issue+Type%3A+Bug+Report%22+-label%3A%22Category%3A+Documentation%22)
  for similar or identical bug reports.
- [x] I have checked the [pull requests list](https://github.com/celery/celery/pulls?q=is%3Apr+label%3A%22PR+Type%3A+Bugfix%22+-label%3A%22Category%3A+Documentation%22)
  for existing proposed fixes.
- [ ] I have checked the [commit log](https://github.com/celery/celery/commits/master)
  to find out if the bug was already fixed in the master branch.
- [x] I have included all related issues and possible duplicate issues
  in this issue (If there are none, check this box anyway).

## Mandatory Debugging Information

- [ ] I have included the output of ``celery -A proj report`` in the issue.
    (if you are not able to do this, then at least specify the Celery
     version affected).
- [ ] I have verified that the issue exists against the `master` branch of Celery.
- [ ] I have included the contents of ``pip freeze`` in the issue.
- [ ] I have included all the versions of all the external dependencies required
  to reproduce this bug.

## Optional Debugging Information
- [ ] I have tried reproducing the issue on more than one Python version
  and/or implementation.
- [ ] I have tried reproducing the issue on more than one message broker and/or
  result backend.
- [ ] I have tried reproducing the issue on more than one version of the message
  broker and/or result backend.
- [ ] I have tried reproducing the issue on more than one operating system.
- [ ] I have tried reproducing the issue on more than one workers pool.
- [ ] I have tried reproducing the issue with autoscaling, retries,
  ETA/Countdown & rate limits disabled.
- [ ] I have tried reproducing the issue after downgrading
  and/or upgrading Celery and its dependencies.

## Environment & Settings
**Celery version**: 5.1.2 (sun-harmonics)

# Steps to Reproduce

Execute `celery -A proj inspect -j active`

# Expected Behavior
Valid JSON string is returned, like: 
```
{"worker-1": [], "worker-2": [], "worker-3": []}
```

# Actual Behavior
Command returns valid JSON in the first line, followed by double newline and some useless summary:
```
{"worker-1": [], "worker-2": [], "worker-3": []}

3 nodes online.
```
which makes the overall output an invalid JSON string.
