/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.samza.container.host;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

/**
 * An implementation of {@link SystemStatisticsGetter} that relies on using Posix commands like ps.
 */
public class PosixCommandBasedStatisticsGetter implements SystemStatisticsGetter {

  private static final Logger log = LoggerFactory.getLogger(PosixCommandBasedStatisticsGetter.class);

  /**
   * A convenience method to execute shell commands and return the first line of their output.
   *
   * @param cmdArray the command to run
   * @return the first line of the output.
   * @throws IOException
   */
  private String getCommandOutput(String[] cmdArray) throws IOException {
    Process executable = Runtime.getRuntime().exec(cmdArray);
    BufferedReader processReader = null;
    String psOutput = null;

    try {
      processReader = new BufferedReader(new InputStreamReader(executable.getInputStream()));
      psOutput = processReader.readLine();
    } finally {
      if (processReader != null) {
        processReader.close();
      }
    }
    return psOutput;
  }

  private long getPhysicalMemory() throws IOException {

    // returns a single long value that represents the rss memory of the process.
    String commandOutput = getCommandOutput(new String[]{"sh", "-c", "ps -o rss= -p $PPID"});

    // this should never happen.
    if (commandOutput == null) {
      throw new IOException("ps returned unexpected output: " + commandOutput);
    }

    long rssMemoryKb = Long.parseLong(commandOutput.trim());
    //convert to bytes
    return rssMemoryKb * 1024;
  }


  @Override
  public SystemMemoryStatistics getSystemMemoryStatistics() {
    try {
      long memory = getPhysicalMemory();
      return new SystemMemoryStatistics(memory);
    } catch (Exception e) {
      log.warn("Error when running ps: ", e);
      return null;
    }
  }
}

@@ -843,16 +843,120 @@
 .host;%0A%0A
+import java.util.List;%0Aimport java.util.stream.Collectors;%0Aimport org.apache.commons.lang3.StringUtils;%0A
 import o
@@ -2095,62 +2095,1356 @@
 %0A%0A  
-private long getPhysicalMemory() throws IOException %7B%0A
+/**%0A   * A convenience method to execute shell commands and return all lines of their output.%0A   *%0A   * @param cmdArray the command to run%0A   * @return all lines of the output.%0A   * @throws IOException%0A   */%0A  private List%3CString%3E getAllCommandOutput(String%5B%5D cmdArray) throws IOException %7B%0A    Process executable = Runtime.getRuntime().exec(cmdArray);%0A    BufferedReader processReader = null;%0A    List%3CString%3E psOutput;%0A%0A    try %7B%0A      processReader = new BufferedReader(new InputStreamReader(executable.getInputStream()));%0A      psOutput = processReader.lines().filter(StringUtils::isNotEmpty).collect(Collectors.toList());%0A    %7D finally %7B%0A      if (processReader != null) %7B%0A        processReader.close();%0A      %7D%0A    %7D%0A    return psOutput;%0A  %7D%0A%0A  private long getTotalPhysicalMemory() throws IOException %7B%0A    // collect all child process ids of the main process that runs the application%0A    List%3CString%3E processIds = getAllCommandOutput(new String%5B%5D%7B%22sh%22, %22-c%22, %22pgrep -P $PPID%22%7D);%0A    // add the parent process which is the main process that runs the application%0A    processIds.add(%22$PPID%22);%0A    long totalPhysicalMemory = 0;%0A    for (String processId : processIds) %7B%0A      totalPhysicalMemory += getPhysicalMemory(processId);%0A    %7D%0A    return totalPhysicalMemory;%0A  %7D%0A%0A  private long getPhysicalMemory(String processId) throws IOException %7B
 %0A   
@@ -3591,16 +3591,30 @@
 , %22-c%22, 
+String.format(
 %22ps -o r
@@ -3620,22 +3620,31 @@
 rss= -p 
-$PPID%22
+%25s%22, processId)
 %7D);%0A%0A   
@@ -4017,24 +4017,29 @@
 memory = get
+Total
 PhysicalMemo

Include memory usage of child processes in physical-memory-mb metric (#1530)