Fix for missing physical plan in UI

examples/src/java/org/apache/heron/examples/api/AckingTopology.java

@@ -20,6 +20,7 @@
 
 package org.apache.heron.examples.api;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Random;
 
@@ -36,6 +37,7 @@
 import org.apache.heron.api.tuple.Fields;
 import org.apache.heron.api.tuple.Tuple;
 import org.apache.heron.api.tuple.Values;
+import org.apache.heron.common.basics.SysUtils;
 
 /**
  * This is a basic example of a Heron topology with acking enable.
@@ -53,7 +55,7 @@ public static void main(String[] args) throws Exception {
 
     int spouts = 2;
     int bolts = 2;
-    builder.setSpout("word", new AckingTestWordSpout(), spouts);
+    builder.setSpout("word", new AckingTestWordSpout(Duration.ofMillis(200)), spouts);
     builder.setBolt("exclaim1", new ExclamationBolt(), bolts)
         .shuffleGrouping("word");
 
@@ -97,8 +99,10 @@ public static class AckingTestWordSpout extends BaseRichSpout {
     private SpoutOutputCollector collector;
     private String[] words;
     private Random rand;
+    private final Duration throttleDuration;
 
-    public AckingTestWordSpout() {
+    public AckingTestWordSpout(Duration throttleDuration) {
+      this.throttleDuration = throttleDuration;
     }
 
     @SuppressWarnings("rawtypes")
@@ -116,7 +120,9 @@ public void close() {
 
     public void nextTuple() {
       final String word = words[rand.nextInt(words.length)];
-
+      if (!throttleDuration.isZero()) {
+        SysUtils.sleep(throttleDuration); // sleep to throttle back CPU usage
+      }
       // To enable acking, we need to emit each tuple with a MessageId, which is an Object.
       // Each new message emitted needs to be annotated with a unique ID, which allows
       // the spout to keep track of which messages should be acked back to the producer or

heron/healthmgr/src/java/org/apache/heron/healthmgr/sensors/TrackerMetricsProvider.java

@@ -63,7 +63,7 @@ public TrackerMetricsProvider(@Named(CONF_METRICS_SOURCE_URL) String trackerURL,
     Client client = ClientBuilder.newClient();
 
     this.baseTarget = client.target(trackerURL)
-        .path("topologies/metricstimeline")
+        .path("topologies/metrics/timeline")
         .queryParam("cluster", cluster)
         .queryParam("environ", environ)
         .queryParam("topology", topologyName);

heron/shell/src/python/handlers/downloadhandler.py

@@ -20,15 +20,14 @@
 
 
 ''' downloadhandler.py '''
-import mimetypes
 import os
 import logging
-from tornado import web, iostream, gen
+import tornado.web
 import anticrlf
 
 from heron.shell.src.python import utils
 
-class DownloadHandler(web.RequestHandler):
+class DownloadHandler(tornado.web.RequestHandler):
   """
   Responsible for downloading the files.
   """
@@ -43,37 +42,41 @@ async def get(self, path):
 
     logger.debug("request to download: %s", path)
 
+    # If the file is large, we want to abandon downloading
+    # if user cancels the requests.
+    # pylint: disable=attribute-defined-outside-init
+    self.connection_closed = False
+
     self.set_header("Content-Disposition", "attachment")
     if not utils.check_path(path):
+      self.write("Only relative paths are allowed")
       self.set_status(403)
-      await self.finish("Only relative paths are allowed")
+      self.finish()
       return
 
     if path is None or not os.path.isfile(path):
+      self.write("File %s  not found" % path)
       self.set_status(404)
-      await self.finish("File %s  not found" % path)
+      self.finish()
       return
 
-    chunk_size = int(4 * 1024 * 1024)
-    content_type = mimetypes.guess_type(path)
-    self.set_header("Content-Type", content_type[0])
-    with open(path, 'rb') as f:
-      while True:
-        chunk = f.read(chunk_size)
-        if not chunk:
-          break
-        try:
-          self.write(chunk)  # write the chunk to response
-          await self.flush() # send the chunk to client
-        except iostream.StreamCloseError:
-          # this means the client has closed the connection
-          # so break the loop
-          break
-        finally:
-          # deleting the chunk is very important because
-          # if many client are downloading files at the
-          # same time, the chunks in memory will keep
-          # increasing and will eat up the RAM
-          del chunk
-          # pause the coroutine so other handlers can run
-          await gen.sleep(0.000000001) # 1 nanosecond
+    length = int(4 * 1024 * 1024)
+    offset = int(0)
+    while True:
+      data = await utils.read_chunk(path, offset=offset, length=length, escape_data=False)
+      if self.connection_closed or 'data' not in data or len(data['data']) < length:
+        break
+      offset += length
+      self.write(data['data'])
+      self.flush()
+
+    if 'data' in data:
+      self.write(data['data'])
+    self.finish()
+
+def on_connection_close(self):
+  '''
+  :return:
+  '''
+  # pylint: disable=attribute-defined-outside-init
+  self.connection_closed = True

heron/shell/src/python/handlers/pidhandler.py

@@ -22,6 +22,7 @@
 ''' pidhandler.py '''
 import json
 import tornado.web
+import subprocess
 
 from heron.shell.src.python import utils
 
@@ -33,5 +34,9 @@ class PidHandler(tornado.web.RequestHandler):
   # pylint: disable=attribute-defined-outside-init
   async def get(self, instance_id):
     ''' get method '''
-    self.content_type = 'application/json'
-    await self.finish(json.dumps(utils.chain([['cat', "%s.pid" % instance_id]])).strip())
+    pid = subprocess.run(['cat', "%s.pid" % instance_id], capture_output=True, text=True)
+    await self.finish({
+      'command': ' '.join(pid.args),
+      'stdout': pid.stdout,
+      'stderr': pid.stderr,
+      })

heron/shell/src/python/main.py

@@ -59,7 +59,7 @@ def run(url_to_handlers=default_handlers):
   AsyncHTTPClient.configure(None, defaults=dict(request_timeout=120.0))
   app = tornado.web.Application(url_to_handlers)
   app.listen(options.port)
-  tornado.ioloop.IOLoop.instance().start()
+  tornado.ioloop.IOLoop.current().start()
 
 if __name__ == '__main__':
   run()

heron/shell/src/python/utils.py

@@ -177,21 +177,6 @@ def str_cmd(cmd, cwd, env):
   stdout, stderr = stdout_builder.result(), stderr_builder.result()
   return {'command': ' '.join(cmd), 'stderr': stderr, 'stdout': stdout}
 
-# pylint: disable=unnecessary-lambda
-def chain(cmd_list):
-  """
-  Feed output of one command to the next and return final output
-  Returns string output of chained application of commands.
-  """
-  command = ' | '.join([' '.join(x) for x in cmd_list])
-  chained_proc = functools.reduce(pipe, [None] + cmd_list)
-  stdout_builder = proc.async_stdout_builder(chained_proc)
-  chained_proc.wait()
-  return {
-      'command': command,
-      'stdout': stdout_builder.result()
-  }
-
 def get_container_id(instance_id):
   ''' get container id '''
   return instance_id.split('_')[1]  # Format: container_<index>_component_name_<index>

heron/tools/common/src/python/clients/tracker.py

@@ -54,8 +54,8 @@
 SCHEDULER_LOCATION_URL_FMT  = "%s/schedulerlocation"  % TOPOLOGIES_URL_FMT
 
 METRICS_URL_FMT             = "%s/metrics"            % TOPOLOGIES_URL_FMT
-METRICS_QUERY_URL_FMT       = "%s/metricsquery"       % TOPOLOGIES_URL_FMT
-METRICS_TIMELINE_URL_FMT    = "%s/metricstimeline"    % TOPOLOGIES_URL_FMT
+METRICS_QUERY_URL_FMT       = "%s/metrics/query"       % TOPOLOGIES_URL_FMT
+METRICS_TIMELINE_URL_FMT    = "%s/metrics/timeline"    % TOPOLOGIES_URL_FMT
 
 EXCEPTIONS_URL_FMT          = "%s/exceptions"         % TOPOLOGIES_URL_FMT
 EXCEPTION_SUMMARY_URL_FMT   = "%s/exceptionsummary"   % TOPOLOGIES_URL_FMT
@@ -66,9 +66,9 @@
 JMAP_URL_FMT                = "%s/jmap"               % TOPOLOGIES_URL_FMT
 HISTOGRAM_URL_FMT           = "%s/histo"              % TOPOLOGIES_URL_FMT
 
-FILE_DATA_URL_FMT           = "%s/containerfiledata"  % TOPOLOGIES_URL_FMT
-FILE_DOWNLOAD_URL_FMT       = "%s/containerfiledownload"  % TOPOLOGIES_URL_FMT
-FILESTATS_URL_FMT           = "%s/containerfilestats" % TOPOLOGIES_URL_FMT
+FILE_DATA_URL_FMT           = "%s/container/filedata"  % TOPOLOGIES_URL_FMT
+FILE_DOWNLOAD_URL_FMT       = "%s/container/filedownload"  % TOPOLOGIES_URL_FMT
+FILESTATS_URL_FMT           = "%s/container/filestats" % TOPOLOGIES_URL_FMT
 
 
 def strip_whitespace(s):
@@ -122,27 +122,26 @@ def strip_whitespace(s):
     backpressure=backpressure
 )
 
-def api_get(url: str, params=None) -> dict:
+def api_get(url: str, params=None) -> Any:
   """Make a GET request to a tracker URL and return the result."""
   start = time.time()
   try:
+    Log.debug(f"Requesting URL: {url} with params: {params}")
     response = requests.get(url, params)
     response.raise_for_status()
   except Exception as e:
-    Log.error(f"Unable to get response from {url}: {e}")
+    Log.error(f"Unable to get response from {url} with params {params}: {e}")
     return None
   end = time.time()
   data = response.json()
-  if data["status"] != "success":
-    Log.error("error from tracker: %s", data["message"])
+  if response.status_code != requests.codes.ok:
+    Log.error("error from tracker: %s", response.status_code)
     return None
 
-  execution = data["executiontime"] * 1000
   duration = (end - start) * 1000
-  Log.debug(f"URL fetch took {execution:.2}ms server time for {url}")
   Log.debug(f"URL fetch took {duration:.2}ms round trip time for {url}")
 
-  return data["result"]
+  return data
 
 
 def create_url(fmt: str) -> str:

heron/tools/tracker/src/python/BUILD

@@ -8,7 +8,7 @@ pex_library(
     ),
     reqs = [
         "click==7.1.2",
-        "fastapi==0.62.0",
+        "fastapi==0.75.0",
         "httpx==0.16.1",
         "javaobj-py3==0.4.1",
         "networkx==2.5",

heron/tools/tracker/src/python/app.py

@@ -26,7 +26,6 @@
 from typing import Dict, List, Optional
 
 from heron.tools.tracker.src.python import constants, state, query
-from heron.tools.tracker.src.python.utils import ResponseEnvelope
 from heron.tools.tracker.src.python.routers import topologies, container, metrics
 
 from fastapi import FastAPI, Query
@@ -80,35 +79,23 @@ async def shutdown_event():
   """Stop recieving topology updates."""
   state.tracker.stop_sync()
 
-
 @app.exception_handler(Exception)
 async def handle_exception(_, exc: Exception):
-  payload = ResponseEnvelope[str](
-      result=None,
-      execution_time=0.0,
-      message=f"request failed: {exc}",
-      status=constants.RESPONSE_STATUS_FAILURE
-  )
+  message=f"request failed: {exc}"
   status_code = 500
   if isinstance(exc, StarletteHTTPException):
     status_code = exc.status_code
   if isinstance(exc, RequestValidationError):
     status_code = 400
-  return JSONResponse(content=payload.dict(), status_code=status_code)
-
+  return JSONResponse(content=message, status_code=status_code)
 
-@app.get("/clusters", response_model=ResponseEnvelope[List[str]])
+@app.get("/clusters")
 async def clusters() -> List[str]:
-  return ResponseEnvelope[List[str]](
-      execution_time=0.0,
-      message="ok",
-      status="success",
-      result=[s.name for s in state.tracker.state_managers],
-  )
-
+  return (s.name for s in state.tracker.state_managers)
+
 @app.get(
     "/machines",
-    response_model=ResponseEnvelope[Dict[str, Dict[str, Dict[str, List[str]]]]],
+    response_model=Dict[str, Dict[str, Dict[str, List[str]]]],
 )
 async def get_machines(
     cluster_names: Optional[List[str]] = Query(None, alias="cluster"),
@@ -134,9 +121,4 @@ async def get_machines(
         topology.name
     ] = topology.get_machines()
 
-  return ResponseEnvelope[Dict[str, Dict[str, Dict[str, List[str]]]]](
-      execution_time=0.0,
-      result=response,
-      status="success",
-      message="ok",
-  )
+  return response

heron/tools/tracker/src/python/main.py

@@ -37,7 +37,7 @@
 import uvicorn
 
 Log = log.Log
-
+Log.setLevel(logging.DEBUG)
 
 def create_tracker_config(config_file: str, stmgr_override: dict) -> dict:
   # try to parse the config file if we find one
@@ -125,6 +125,7 @@ def cli(
 
   log_level = logging.DEBUG if verbose else logging.INFO
   log.configure(log_level)
+  Log = log.Log
 
   stmgr_override = {
       "type": stmgr_type,

heron/tools/tracker/src/python/metricstimeline.py

@@ -32,9 +32,9 @@
 
 class MetricsTimeline(BaseModel):
   component: str
-  start_time: int = Field(..., alias="starttime")
-  end_time: int = Field(..., alias="enddtime")
-  timeline: Dict[str, Dict[str, Dict[int, int]]] = Field(
+  starttime: int = Field(..., alias="starttime")
+  endtime: int = Field(..., alias="endtime")
+  timeline: Dict[str, Dict[str, Dict[int, str]]] = Field(
       ...,
       description="map of (metric name, instance, start) to metric value",
   )
@@ -73,21 +73,21 @@ async def get_metrics_timeline(
 
   # Form and send the http request.
   url = f"http://{tmanager.host}:{tmanager.stats_port}/stats"
-  with httpx.AsyncClient() as client:
+  async with httpx.AsyncClient() as client:
     result = await client.post(url, data=request_parameters.SerializeToString())
 
   # Check the response code - error if it is in 400s or 500s
-  if result.code >= 400:
+  if result.status_code >= 400:
     message = f"Error in getting metrics from Tmanager, code: {result.code}"
     raise Exception(message)
 
   # Parse the response from tmanager.
   response_data = tmanager_pb2.MetricResponse()
   response_data.ParseFromString(result.content)
 
-  if response_data.status.status == common_pb2.NOTOK:
-    if response_data.status.HasField("message"):
-      Log.warn("Received response from Tmanager: %s", response_data.status.message)
+  # if response_data.status.status == common_pb2.NOTOK:
+  #   if response_data.status.HasField("message"):
+  #     Log.warn("Received response from Tmanager: %s", response_data.status.message)
 
   timeline = {}
   # Loop through all the metrics
@@ -99,6 +99,8 @@ async def get_metrics_timeline(
     # Loop through all individual metrics.
     for im in metric.metric:
       metricname = im.name
+      if metricname not in timeline:
+        timeline.setdefault(metricname, {})
       if instance not in timeline[metricname]:
         timeline.setdefault(metricname, {})[instance] = {}