# ignite-690 WIP
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/82733d41 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/82733d41 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/82733d41 Branch: refs/heads/ignite-sprint-4 Commit: 82733d412f199b5fe588d3810b11da8331a25b9e Parents: e8a9dfb Author: Andrey <anovi...@gridgain.com> Authored: Thu Apr 16 15:26:53 2015 +0700 Committer: Andrey <anovi...@gridgain.com> Committed: Thu Apr 16 15:26:53 2015 +0700 ---------------------------------------------------------------------- .../visor/commands/ack/VisorAckCommand.scala | 11 ++- .../visor/commands/gc/VisorGcCommand.scala | 6 +- .../commands/tasks/VisorTasksCommand.scala | 22 ++--- .../scala/org/apache/ignite/visor/visor.scala | 91 ++++++++++++++------ 4 files changed, 86 insertions(+), 44 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/82733d41/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/ack/VisorAckCommand.scala ---------------------------------------------------------------------- diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/ack/VisorAckCommand.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/ack/VisorAckCommand.scala index 6fa96d0..6faa276 100644 --- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/ack/VisorAckCommand.scala +++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/ack/VisorAckCommand.scala @@ -18,15 +18,14 @@ package org.apache.ignite.visor.commands.ack import org.apache.ignite.cluster.ClusterGroupEmptyException -import org.apache.ignite.internal.visor.misc.VisorAckTask +import org.apache.ignite.visor.VisorTag +import org.apache.ignite.visor.commands.VisorConsoleCommand +import org.apache.ignite.visor.visor._ import java.util.{HashSet => JavaHashSet} -import org.apache.ignite.visor.commands.VisorConsoleCommand -import org.apache.ignite.visor.visor._ -import org.apache.ignite.visor.{VisorTag, visor} +import org.apache.ignite.internal.visor.misc.VisorAckTask -import scala.collection.JavaConversions._ import scala.language.implicitConversions /** @@ -105,7 +104,7 @@ class VisorAckCommand { adviseToConnect() else try { - executeAll(classOf[VisorAckTask], msg) + executeMulti(classOf[VisorAckTask], msg) } catch { case _: ClusterGroupEmptyException => scold("Topology is empty.") http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/82733d41/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/gc/VisorGcCommand.scala ---------------------------------------------------------------------- diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/gc/VisorGcCommand.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/gc/VisorGcCommand.scala index 1c15528..9b5fc21 100644 --- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/gc/VisorGcCommand.scala +++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/gc/VisorGcCommand.scala @@ -18,15 +18,15 @@ package org.apache.ignite.visor.commands.gc import org.apache.ignite._ - import org.apache.ignite.cluster.{ClusterGroupEmptyException, ClusterNode} -import org.apache.ignite.internal.visor.node.VisorNodeGcTask import org.apache.ignite.visor.VisorTag import org.apache.ignite.visor.commands.{VisorConsoleCommand, VisorTextTable} import org.apache.ignite.visor.visor._ import java.util.UUID +import org.apache.ignite.internal.visor.node.VisorNodeGcTask + import scala.collection.JavaConversions._ import scala.language.{implicitConversions, reflectiveCalls} import scala.util.control.Breaks._ @@ -143,7 +143,7 @@ class VisorGcCommand { val NULL: Void = null - executeRemotes(classOf[VisorNodeGcTask], NULL).foreach { + executeMulti(classOf[VisorNodeGcTask], NULL).foreach { case (nid, stat) => val roundHb = stat.get1() / (1024L * 1024L) val roundHa = stat.get2() / (1024L * 1024L) http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/82733d41/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/tasks/VisorTasksCommand.scala ---------------------------------------------------------------------- diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/tasks/VisorTasksCommand.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/tasks/VisorTasksCommand.scala index 515864d..d7547c9 100644 --- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/tasks/VisorTasksCommand.scala +++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/commands/tasks/VisorTasksCommand.scala @@ -21,17 +21,17 @@ import org.apache.ignite._ import org.apache.ignite.events.EventType._ import org.apache.ignite.internal.util.typedef.X import org.apache.ignite.internal.util.{IgniteUtils => U} -import org.apache.ignite.internal.visor.event.{VisorGridEvent, VisorGridJobEvent, VisorGridTaskEvent} -import org.apache.ignite.internal.visor.node.VisorNodeEventsCollectorTask -import org.apache.ignite.internal.visor.node.VisorNodeEventsCollectorTask.VisorNodeEventsCollectorTaskArg import org.apache.ignite.lang.IgniteUuid - -import java.util.UUID - import org.apache.ignite.visor.VisorTag import org.apache.ignite.visor.commands.{VisorConsoleCommand, VisorTextTable} import org.apache.ignite.visor.visor._ +import java.util.UUID + +import org.apache.ignite.internal.visor.event.{VisorGridEvent, VisorGridJobEvent, VisorGridTaskEvent} +import org.apache.ignite.internal.visor.node.VisorNodeEventsCollectorTask +import org.apache.ignite.internal.visor.node.VisorNodeEventsCollectorTask.VisorNodeEventsCollectorTaskArg + import scala.collection.JavaConversions._ import scala.language.implicitConversions import scala.util.control.Breaks._ @@ -613,7 +613,7 @@ class VisorTasksCommand { private def list(p: Long, taskName: String, reverse: Boolean, all: Boolean) { breakable { try { - val evts = executeRemotes(classOf[VisorNodeEventsCollectorTask], + val evts = executeMulti(classOf[VisorNodeEventsCollectorTask], VisorNodeEventsCollectorTaskArg.createTasksArg(p, taskName, null)) val (tLst, eLst) = mkData(evts) @@ -817,7 +817,7 @@ class VisorTasksCommand { try { val prj = ignite.cluster.forRemotes() - val evts = executeRemotes(classOf[VisorNodeEventsCollectorTask], + val evts = executeMulti(classOf[VisorNodeEventsCollectorTask], VisorNodeEventsCollectorTaskArg.createTasksArg(null, taskName, null)) val (tLst, eLst) = mkData(evts) @@ -988,7 +988,7 @@ class VisorTasksCommand { } try { - val evts = executeRemotes(classOf[VisorNodeEventsCollectorTask], + val evts = executeMulti(classOf[VisorNodeEventsCollectorTask], VisorNodeEventsCollectorTaskArg.createTasksArg(null, null, uuid)) val (tLst, eLst) = mkData(evts) @@ -1100,7 +1100,7 @@ class VisorTasksCommand { private def nodes(f: Long) { breakable { try { - val evts = executeRemotes(classOf[VisorNodeEventsCollectorTask], + val evts = executeMulti(classOf[VisorNodeEventsCollectorTask], VisorNodeEventsCollectorTaskArg.createTasksArg(f, null, null)) val eLst = mkData(evts)._2 @@ -1210,7 +1210,7 @@ class VisorTasksCommand { private def hosts(f: Long) { breakable { try { - val evts = executeRemotes(classOf[VisorNodeEventsCollectorTask], + val evts = executeMulti(classOf[VisorNodeEventsCollectorTask], VisorNodeEventsCollectorTaskArg.createTasksArg(f, null, null)) val eLst = mkData(evts)._2 http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/82733d41/modules/visor-console/src/main/scala/org/apache/ignite/visor/visor.scala ---------------------------------------------------------------------- diff --git a/modules/visor-console/src/main/scala/org/apache/ignite/visor/visor.scala b/modules/visor-console/src/main/scala/org/apache/ignite/visor/visor.scala index 431701a..321d02e 100644 --- a/modules/visor-console/src/main/scala/org/apache/ignite/visor/visor.scala +++ b/modules/visor-console/src/main/scala/org/apache/ignite/visor/visor.scala @@ -19,7 +19,7 @@ package org.apache.ignite.visor import org.apache.ignite.IgniteSystemProperties._ import org.apache.ignite._ -import org.apache.ignite.cluster.{ClusterGroupEmptyException, ClusterGroup, ClusterMetrics, ClusterNode} +import org.apache.ignite.cluster.{ClusterGroup, ClusterGroupEmptyException, ClusterMetrics, ClusterNode} import org.apache.ignite.configuration.IgniteConfiguration import org.apache.ignite.events.EventType._ import org.apache.ignite.events.{DiscoveryEvent, Event} @@ -29,19 +29,16 @@ import org.apache.ignite.internal.IgniteNodeAttributes._ import org.apache.ignite.internal.IgniteVersionUtils._ import org.apache.ignite.internal.cluster.ClusterGroupEmptyCheckedException import org.apache.ignite.internal.util.lang.{GridFunc => F} +import org.apache.ignite.internal.util.spring.IgniteSpringHelper import org.apache.ignite.internal.util.typedef._ import org.apache.ignite.internal.util.{GridConfigurationFinder, IgniteUtils => U} -import org.apache.ignite.logger.NullLogger -import org.apache.ignite.internal.visor.{VisorMultiNodeTask, VisorTaskArgument} -import org.apache.ignite.internal.visor.cache._ -import org.apache.ignite.internal.visor.node._ -import org.apache.ignite.internal.visor.node.VisorNodeEventsCollectorTask.VisorNodeEventsCollectorTaskArg -import org.apache.ignite.internal.visor.util.VisorTaskUtils._ import org.apache.ignite.lang._ +import org.apache.ignite.logger.NullLogger import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi import org.apache.ignite.thread.IgniteThreadPoolExecutor import org.apache.ignite.visor.commands.VisorConsole.consoleReader import org.apache.ignite.visor.commands.{VisorConsoleCommand, VisorTextTable} + import org.jetbrains.annotations.Nullable import java.io._ @@ -50,11 +47,17 @@ import java.text._ import java.util.concurrent._ import java.util.{Collection => JavaCollection, HashSet => JavaHashSet, _} +import org.apache.ignite.internal.visor.cache._ +import org.apache.ignite.internal.visor.node.VisorNodeEventsCollectorTask.VisorNodeEventsCollectorTaskArg +import org.apache.ignite.internal.visor.node._ +import org.apache.ignite.internal.visor.util.VisorEventMapper +import org.apache.ignite.internal.visor.util.VisorTaskUtils._ +import org.apache.ignite.internal.visor.{VisorMultiNodeTask, VisorTaskArgument} + import scala.collection.JavaConversions._ import scala.collection.immutable import scala.language.{implicitConversions, reflectiveCalls} import scala.util.control.Breaks._ -import org.apache.ignite.internal.util.spring.IgniteSpringHelper /** * Holder for command help information. @@ -1848,24 +1851,60 @@ object visor extends VisorTag { ignite.compute(grp).withNoFailover().execute(task, toTaskArgument(grp.nodes().map(_.id()), arg)) } + /** + * Execute task on node. + * + * @param nid Node id. + * @param task Task class + * @param arg Task argument. + * @tparam A Task argument type. + * @tparam R Task result type + * @tparam J Job class. + * @return Task result. + */ def executeOne[A, R, J](nid: UUID, task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = execute(ignite.cluster.forNodeId(nid), task, arg) + /** + * Execute task on random node. + * + * @param task Task class + * @param arg Task argument. + * @tparam A Task argument type. + * @tparam R Task result type + * @tparam J Job class. + * @return Task result. + */ + def executeRandom[A, R, J](task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = + execute(ignite.cluster.forRandom(), task, arg) + + /** + * Execute task on specified nodes. + * + * @param nids Node ids. + * @param task Task class + * @param arg Task argument. + * @tparam A Task argument type. + * @tparam R Task result type + * @tparam J Job class. + * @return Task result. + */ def executeMulti[A, R, J](nids: Iterable[UUID], task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = execute(ignite.cluster.forNodeIds(nids), task, arg) - def executeLocal[A, R, J](task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = - execute(ignite.cluster.forLocal(), task, arg) - - def executeRemotes[A, R, J](task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = + /** + * Execute task on all nodes. + * + * @param task Task class + * @param arg Task argument. + * @tparam A Task argument type. + * @tparam R Task result type + * @tparam J Job class. + * @return Task result. + */ + def executeMulti[A, R, J](task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = execute(ignite.cluster.forRemotes(), task, arg) - def executeAll[A, R, J](task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = - execute(ignite.cluster, task, arg) - - def executeRandom[A, R, J](task: Class[_ <: VisorMultiNodeTask[A, R, J]], arg: A) = - execute(ignite.cluster.forRandom(), task, arg) - /** * Gets configuration from specified node. * @@ -2393,6 +2432,12 @@ object visor extends VisorTag { println("<visor>: Log stopped: " + logFile.getAbsolutePath) } + /** Unique Visor key to get events last order. */ + final val EVT_LAST_ORDER_KEY = UUID.randomUUID().toString + + /** Unique Visor key to get events throttle counter. */ + final val EVT_THROTTLE_CNTR_KEY = UUID.randomUUID().toString + /** * Starts logging. If logging is already started - no-op. * @@ -2477,16 +2522,14 @@ object visor extends VisorTag { ) override def run() { - val g = ignite - - if (g != null) { + if (ignite != null) { try { // Discovery events collected only locally. - val loc = executeLocal(classOf[VisorNodeEventsCollectorTask], - VisorNodeEventsCollectorTaskArg.createLogArg(key, LOG_EVTS ++ EVTS_DISCOVERY)).toSeq + val loc = collectEvents(ignite, EVT_LAST_ORDER_KEY, EVT_THROTTLE_CNTR_KEY, + LOG_EVTS ++ EVTS_DISCOVERY, new VisorEventMapper).toSeq val evts = if (!rmtLogDisabled) - loc ++ executeRemotes(classOf[VisorNodeEventsCollectorTask], + loc ++ executeMulti(classOf[VisorNodeEventsCollectorTask], VisorNodeEventsCollectorTaskArg.createLogArg(key, LOG_EVTS)).toSeq else loc
