[dashboard] fix peek parse message error (apache#4918)

Fixes apache#4917 

### Motivation

dashboard peek message api raise exception when format response cause treat all message as a JSON string

### Modifications

* format as JSON only if message is a JSON, 
* otherwise if message is printable, return the original message,
* otherwise print hex like command with --hex option.

dashboard/Dockerfile

@@ -25,7 +25,7 @@ RUN apt-get update
 RUN apt-get -y install postgresql python sudo nginx supervisor
 
 # Python dependencies
-RUN pip install uwsgi 'Django<2.0' psycopg2 pytz requests
+RUN pip install uwsgi 'Django<2.0' psycopg2 pytz requests hexdump
 
 # Postgres configuration
 COPY conf/postgresql.conf /etc/postgresql/9.6/main/

dashboard/django/stats/templates/stats/peek.html

@@ -19,4 +19,7 @@
 
 -->
 
-<pre id="output" class="autoscroll">{{ message_body }}</pre>
+<div id="output" class="autoscroll">
+    <pre>{{ message_type }}</pre>
+    <pre>{{ message_body }}</pre>
+</div>

dashboard/django/stats/views.py

@@ -18,12 +18,15 @@
 #
 
 import logging
+import struct
+
 from django.shortcuts import render, get_object_or_404, redirect
 from django.template import loader
 from django.urls import reverse
 from django.views import generic
 from django.db.models import Q, IntegerField
 from dashboard import settings
+import hexdump
 import requests, json, re
 
 from django.http import HttpResponseRedirect, HttpResponse
@@ -363,13 +366,51 @@ def messages(request, topic_name, subscription_name):
         'subtitle' : subscription_name,
     })
 
+
+def message_skip_meta(message_view):
+    if not message_view or len(message_view) < 4:
+        raise ValueError("invalid message")
+    meta_size = struct.unpack(">I", message_view[:4])
+    message_index = 4 + meta_size[0]
+    if len(message_view) < message_index:
+        raise ValueError("invalid message")
+    return message_view[message_index:]
+
+
+def get_message_from_http_response(response):
+    if response.status_code != 200:
+        return "ERROR", "status_code=%d" % response.status_code
+    message_view = memoryview(response.content)
+    if 'X-Pulsar-num-batch-message' in response.headers:
+        batch_size = int(response.headers['X-Pulsar-num-batch-message'])
+        if batch_size == 1:
+            message_view = message_skip_meta(message_view)
+        else:
+            # TODO: can not figure out multi-message batch for now
+            return "Batch(size=%d)" % batch_size, "<omitted>"
+
+    try:
+        text = str(message_view,
+                   encoding=response.encoding or response.apparent_encoding,
+                   errors='replace')
+        if not text.isprintable():
+            return "Hex", hexdump.hexdump(message_view, result='return')
+    except (LookupError, TypeError):
+        return "Hex", hexdump.hexdump(message_view, result='return')
+    try:
+        return "JSON", json.dumps(json.loads(text),
+                                  ensure_ascii=False, indent=4)
+    except json.JSONDecodeError:
+        return "Text", text
+
+
 def peek(request, topic_name, subscription_name, message_number):
     url = settings.SERVICE_URL + '/admin/v2/' + topic_name + '/subscription/' + subscription_name + '/position/' + message_number
     response = requests.get(url)
-    message = response.text
-    message = message[message.index('{'):]
+    message_type, message = get_message_from_http_response(response)
     context = {
-        'message_body' : json.dumps(json.loads(message), indent=4),
+        'message_type': message_type,
+        'message_body': message,
     }
     return render(request, 'stats/peek.html', context)