[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/553 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user tillrohrmann commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-100220928 LGTM :-) --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-100221773 Great. Once this is green, I'll push it to master: https://travis-ci.org/rmetzger/flink/builds/61757934 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-99786510 Hey Robert, please correct me if I am wrong. According to this doc, https://docs.oracle.com/javase/7/docs/jre/api/management/extension/com/sun/management/OperatingSystemMXBean.html OperatingSystemMXBean of the sun package is available for Java 6 too. Only the method getProcessCpuTime is unavailable. Hence I believe the logging and exception will not occur for each heartbeat. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/553#issuecomment-99759058
That would be more javaesque than scalaesque. I would rather move the
osMXBean retrieval into the map function of the `fetchCPULoad` option. That
also solves the problem.
```
fetchCPULoad.map{
method =
val osMXBean =
method.invoke(osMXBean).asInstanceOf[Double]
}.getOrElse(-1)
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-99787426 Okay, you are right. Then we can keep it this way. Let me try out your changes one more time, but then I think we are good to merge. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/553#issuecomment-99753925
The code looks much better now. The only thing that makes me still unhappy
is the following method
```diff
+ override def getValue: Double = {
+try{
+ val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+asInstanceOf[com.sun.management.OperatingSystemMXBean]
+
fetchCPULoad.map(_.invoke(osMXBean).asInstanceOf[Double]).getOrElse(-1)
+} catch {
+ case t: Throwable = {
+LOG.warn(Error retrieving CPU Load through
OperatingSystemMXBean, t)
+-1
+ }
+}
+ }
```
This `getValue()` method is called every 5 seconds to get the metrics from
each machine. Users using Java 6 will get the WARN in their log + the exception
because the cast to `com.sun.management.OperatingSystemMXBean` will fail every
time.
I would do the following: If you detect that fetchCpuLoad is null, register
a gauge which is always returning -1. Otherwise, register the gauge which is
calling fetchCpuLoad.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-99804027 I have made the change. Thanks a lot Robert and Till. I too am looking forward to contribute more :) --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29744552
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,49 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+// Preprocessing steps for registering cpuLoad
+// dummy initialisation
+var fetchCPULoad:(Any) = Double = (obj:Any) = -1
+
+// define the fetchCPULoad method as per the fetched getProcessCpuLoad
method
+getMethodToFetchCPULoad() match {
+ case Some(method) = fetchCPULoad = (obj:Any) =
method.asInstanceOf[Method].invoke(obj).
+asInstanceOf[Double]
+ // Log getProcessCpuLoad method not available for Java 6
+ case None = LOG.warn(getProcessCpuLoad method not available in the
Operating System Bean +
+implementation for this Java runtime environment\n +
Thread.currentThread().getStackTrace)
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
+ val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+asInstanceOf[com.sun.management.OperatingSystemMXBean]
+ fetchCPULoad(osMXBean)
+} catch {
+ case t: Throwable = {
+LOG.warn(Error retrieving CPU Load through
OperatingSystemMXBean, t)
+-1
+ }
+}
+ }
+})
metricRegistry
}
+
+ /**
+ * Fetches getProcessCpuLoad method if available in the
+ * OperatingSystemMXBean implementation else returns None
+ * @return
+ */
+ private def getMethodToFetchCPULoad(): Option[Method] = {
+val methodsList =
classOf[com.sun.management.OperatingSystemMXBean].getMethods()
+val method = methodsList.filter(_.getName == getProcessCpuLoad)
--- End diff --
Shortcut: use `headOption` to obtain result.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29745454
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// Log getProcessCpuLoad method not available for Java 6
+if(getCPULoadMethod == null){
+ LOG.warn(getProcessCpuLoad method not available in the Operating
System Bean +
+ implementation for this Java runtime
environment\n+Thread.currentThread().getStackTrace)
+}
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+ (obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+} else {
+ (obj: Any) = -1
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
--- End diff --
why not doing:
```
val fetchCPULoad = getMethodToFetchCPULoad()
if(fetchCPULoad.isEmpty) { LOG.warn(Java 6 not supported ...) }
metricRegistry.register(cpuLoad, new Gauge[Double] {
override def getValue(): Double = {
try{
val osMXBean = ManagementFactory.getOperatinSystemMXBean()
fetchCPULoad.map(_.invoke(osMXBean))
}.getOrElse(-1)
} catch {
case t: Throwable = {
LOG.warn()
-1
}
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29786547
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// Log getProcessCpuLoad method not available for Java 6
+if(getCPULoadMethod == null){
+ LOG.warn(getProcessCpuLoad method not available in the Operating
System Bean +
+ implementation for this Java runtime
environment\n+Thread.currentThread().getStackTrace)
+}
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+ (obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+} else {
+ (obj: Any) = -1
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
--- End diff --
Awesome :). These features in Scala are just too good :D
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29662593
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+// dummy initialisation
+var fetchCPULoad:(Any) = Double = (obj:Any) = -1
+
+if(getCPULoadMethod != null){
+ fetchCPULoad = (obj:Any) =
getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+} else {
+ // Log getProcessCpuLoad method not available for Java 6
+ LOG.warn(getProcessCpuLoad method not available in the Operating
System Bean +
+implementation for this Java runtime environment\n +
Thread.currentThread().getStackTrace)
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
+ val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+asInstanceOf[com.sun.management.OperatingSystemMXBean]
+ return fetchCPULoad(osMXBean)
--- End diff --
no return statement necessary in Scala
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29662751
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
--- End diff --
Why not making `getCPULoadMethod` an `Option[Method]`? That way one avoids
vicious `null` values.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29682706
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+// dummy initialisation
+var fetchCPULoad:(Any) = Double = (obj:Any) = -1
--- End diff --
We need to define fetchCPULoad method as per the object returned by
getMethodToFetchCPULoad. val cannot be reassigned and has to be initialised.
Hence var has been used to overcome this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29648339
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// Log getProcessCpuLoad method not available for Java 6
+if(getCPULoadMethod == null){
+ LOG.warn(getProcessCpuLoad method not available in the Operating
System Bean +
+ implementation for this Java runtime
environment\n+Thread.currentThread().getStackTrace)
+}
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+ (obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+} else {
+ (obj: Any) = -1
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
--- End diff --
Why aren't you doing the check `getCPULoadMethod != null` before the cast?
if it is null, return -1, if not, do the cast and fetch the CPU load.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29650229
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// Log getProcessCpuLoad method not available for Java 6
+if(getCPULoadMethod == null){
+ LOG.warn(getProcessCpuLoad method not available in the Operating
System Bean +
+ implementation for this Java runtime
environment\n+Thread.currentThread().getStackTrace)
+}
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+ (obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+} else {
+ (obj: Any) = -1
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
--- End diff --
Hey, correct me if I am wrong. I found that Scala needs to initialise a
variable during declaration. Hence if the variable is declared in an if else
clause, it won't be visible to the rest of the code. And reassignment to the
val was giving errors. Hence I have made use of the null check.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29650540
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -1894,6 +1895,52 @@ object TaskManager {
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+// Preprocessing steps for registering cpuLoad
+// fetch the method to get process CPU load
+val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+// Log getProcessCpuLoad method not available for Java 6
+if(getCPULoadMethod == null){
+ LOG.warn(getProcessCpuLoad method not available in the Operating
System Bean +
+ implementation for this Java runtime
environment\n+Thread.currentThread().getStackTrace)
+}
+
+// define the fetchCPULoad method as per the fetched getCPULoadMethod
+val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+ (obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+} else {
+ (obj: Any) = -1
+}
+
+metricRegistry.register(cpuLoad, new Gauge[Double] {
+ override def getValue: Double = {
+try{
--- End diff --
I will try using var instead of val and update the PR immediately
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29572469
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -129,6 +130,41 @@ class TaskManager(val connectionInfo:
InstanceConnectionInfo,
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+ // Preprocessing steps for registering cpuLoad
+ // fetch the method to get process CPU load
+ val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+ // define the fetchCPULoad method as per the fetched getCPULoadMethod
+ val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+(obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+ } else {
+(obj: Any) = {
+ log.warning(getProcessCpuLoad method not available in the Operating
System Bean +
--- End diff --
Correct me if i'm wrong, but this log message will appear everytime the
cpuLoad is acquired (I think by default 5 seconds).
So Java 6 user's will have quite a polluted log
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29572528
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -129,6 +130,41 @@ class TaskManager(val connectionInfo:
InstanceConnectionInfo,
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+ // Preprocessing steps for registering cpuLoad
+ // fetch the method to get process CPU load
+ val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+ // define the fetchCPULoad method as per the fetched getCPULoadMethod
+ val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+(obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+ } else {
+(obj: Any) = {
+ log.warning(getProcessCpuLoad method not available in the Operating
System Bean +
+implementation for this Java runtime
environment,Thread.currentThread().getStackTrace)
+ -1
+}
+ }
+
+ metricRegistry.register(cpuLoad, new Gauge[Double] {
+override def getValue: Double = {
+ try{
+val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+ asInstanceOf[com.sun.management.OperatingSystemMXBean]
+return fetchCPULoad(osMXBean)
+ } catch {
+case t:Throwable = {
+ if (t.isInstanceOf[java.lang.ClassCastException]){
+log.warning(Error casting to OperatingSystemMXBean,t)
--- End diff --
I think this error will also happen every 5 seconds
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29572548
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -129,6 +130,41 @@ class TaskManager(val connectionInfo:
InstanceConnectionInfo,
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+ // Preprocessing steps for registering cpuLoad
+ // fetch the method to get process CPU load
+ val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+ // define the fetchCPULoad method as per the fetched getCPULoadMethod
+ val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+(obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+ } else {
+(obj: Any) = {
+ log.warning(getProcessCpuLoad method not available in the Operating
System Bean +
+implementation for this Java runtime
environment,Thread.currentThread().getStackTrace)
+ -1
+}
+ }
+
+ metricRegistry.register(cpuLoad, new Gauge[Double] {
+override def getValue: Double = {
+ try{
+val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+ asInstanceOf[com.sun.management.OperatingSystemMXBean]
+return fetchCPULoad(osMXBean)
+ } catch {
+case t:Throwable = {
+ if (t.isInstanceOf[java.lang.ClassCastException]){
--- End diff --
I'm not a Scala expert, but I think you can avoid the t.isInstanceOf by
matching the exception with the pattern matching (case ..)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-98640261 I tried our your changes and looked at the change. - The pull request includes commits which are probably not part of the change. Maybe some merging/rebasing went wrong. Can you try to clean up the branch? Maybe a simple rebase on the current master is sufficient, or you might need to manually cherry-pick your changes on the current master. - The Avg value of the Memory Statistics are not aligned. Probably because the positioning is done using spaces. I think it would be better to use a table there.  - The position of the Show Detailed Graph and the Hide Detailed Graph button is not consistent. On is on the bottom of the table cell, the other on the top. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29636451
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -129,6 +130,41 @@ class TaskManager(val connectionInfo:
InstanceConnectionInfo,
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+ // Preprocessing steps for registering cpuLoad
+ // fetch the method to get process CPU load
+ val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+ // define the fetchCPULoad method as per the fetched getCPULoadMethod
+ val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+(obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+ } else {
+(obj: Any) = {
+ log.warning(getProcessCpuLoad method not available in the Operating
System Bean +
+implementation for this Java runtime
environment,Thread.currentThread().getStackTrace)
+ -1
+}
+ }
+
+ metricRegistry.register(cpuLoad, new Gauge[Double] {
+override def getValue: Double = {
+ try{
+val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+ asInstanceOf[com.sun.management.OperatingSystemMXBean]
+return fetchCPULoad(osMXBean)
+ } catch {
+case t:Throwable = {
+ if (t.isInstanceOf[java.lang.ClassCastException]){
--- End diff --
Agreed :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r29636610
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -129,6 +130,41 @@ class TaskManager(val connectionInfo:
InstanceConnectionInfo,
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+
+ // Preprocessing steps for registering cpuLoad
+ // fetch the method to get process CPU load
+ val getCPULoadMethod: Method = getMethodToFetchCPULoad()
+
+ // define the fetchCPULoad method as per the fetched getCPULoadMethod
+ val fetchCPULoad: (Any) = Double = if (getCPULoadMethod != null) {
+(obj: Any) = getCPULoadMethod.invoke(obj).asInstanceOf[Double]
+ } else {
+(obj: Any) = {
+ log.warning(getProcessCpuLoad method not available in the Operating
System Bean +
+implementation for this Java runtime
environment,Thread.currentThread().getStackTrace)
+ -1
+}
+ }
+
+ metricRegistry.register(cpuLoad, new Gauge[Double] {
+override def getValue: Double = {
+ try{
+val osMXBean = ManagementFactory.getOperatingSystemMXBean().
+ asInstanceOf[com.sun.management.OperatingSystemMXBean]
+return fetchCPULoad(osMXBean)
+ } catch {
+case t:Throwable = {
+ if (t.isInstanceOf[java.lang.ClassCastException]){
+log.warning(Error casting to OperatingSystemMXBean,t)
--- End diff --
No, this error is not occurring for each heartbeat. This is in case there
is a new error while using the getProcessCpuLoad method. I will remove the
ClassCastException and make it Throwable so as to catch all types of errors.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-96631425 Thank you. I'm trying to find time to review your changes soon. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-95480100 The suggested changes have been made --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-89987417 The compilation error has been fixed. CPU stats are loading properly for Java 7 and above. For Java 6, the CPU stats column shows Not Available with a popup message explaining why. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user bhatsachin commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-89848884 Hey Robert, The required changes have been made and are working fine for Java 7 onwards. Using this method, it is not possible to fetch CPU load in Java 6 since there is no implementation of the getProcessCpuLoad method available for OperatingSystemMXBean: http://docs.oracle.com/javase/7/docs/api/java/lang/management/OperatingSystemMXBean.html There is a compilation error due to the unavailable method in jdk 6 and hence the Travis failures. What do you suggest should be the next step? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-88828965 Please let me know when the PR is ready for review again. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/553#discussion_r27551487
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
---
@@ -129,6 +129,10 @@ class TaskManager(val connectionInfo:
InstanceConnectionInfo,
override def getValue: Double =
ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage()
})
+ metricRegistry.register(cpuLoad, new Gauge[Double] {
+override def getValue: Double =
+
ManagementFactory.getOperatingSystemMXBean().asInstanceOf[com.sun.management.OperatingSystemMXBean].getProcessCpuLoad()
--- End diff --
This cast may fail on some JVM implementations. See
http://www.oracle.com/technetwork/java/faq-sun-packages-142232.html
Can you surround the metric registration by a try catch block and log a
warning in case the cast fails?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-88387132 Thank you for the contribution. I'll have a look at the changes. The integration test of your change (https://travis-ci.org/apache/flink/builds/56641563) has failed due to: ``` [INFO] --- scalastyle-maven-plugin:0.5.0:check (default) @ flink-runtime --- error file=/home/travis/build/apache/flink/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala message=File line length exceeds 100 characters line=134 Saving to outputFile=/home/travis/build/apache/flink/flink-runtime/scalastyle-output.xml Processed 45 file(s) ``` To update your pull request, you can just push to the branch this PR is based on. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
Github user rmetzger commented on the pull request: https://github.com/apache/flink/pull/553#issuecomment-88418659 The memory statistics get pretty big when the chart is enabled as well. Can you hide the string-based statistics when the graph is activated? http://i.imgur.com/VE0p3Dm.png Also, an you make the button appear in the same design as the other buttons on that page? (I think you have to use the bootstrap button classes) --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request: [FLINK-1792] TM Monitoring: CPU utilization, h...
GitHub user bhatsachin opened a pull request: https://github.com/apache/flink/pull/553 [FLINK-1792] TM Monitoring: CPU utilization, hide graphs by default and show summary only Added CPU Load %, show/hide button for detailed graph and summary for the metrics You can merge this pull request into a Git repository by running: $ git pull https://github.com/bhatsachin/flink dev-bhatsachin-flink-1792 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/553.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #553 commit 5f5f6a9156a8bbe7dc98a998ade7dc0eab9b2912 Author: bhatsachin bhats...@gmail.com Date: 2015-03-31T08:18:08Z [FLINK-1792] Add processCPULoad in metricsRegistry, add button to show/hide graphs, add summary for metrics --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
