[WIP] Decision making module for automatic migration by AmitRoushan · Pull Request #745 · amino-os/Amino.Run

AmitRoushan · 2019-05-04T15:21:52Z

PR has following changes:

code for metric aggregation from kernel server latency and bandwidth collection
code for metric collection for execution time for specific replica
code for metric aggregation for data exchanged between neighbour replica
class for periodically watch metric and call prediction module for top migration candidate
class for predict kernel server node for replica.

Note:

This PR has commit "7c1e364" which contains code changes related to metric collection [PR-1]Node level metrics[latency and data rate] measurement on Kernel Servers #742 during Object monitoring. Code has been added for integration/testing. Following file has code specific to mentioned commit :

core/src/main/java/amino/run/kernel/client/KernelClient.java 
core/src/main/java/amino/run/kernel/common/ServerInfo.java 
core/src/main/java/amino/run/kernel/common/metric/NodeMetric.java 
core/src/main/java/amino/run/kernel/server/KernelServer.java 
core/src/main/java/amino/run/kernel/server/KernelServerImpl.java 
core/src/main/java/amino/run/oms/KernelServerManager.java 
core/src/main/java/amino/run/oms/OMSServerImpl.java 
core/src/test/java/amino/run/common/BaseTest.java 
core/src/test/java/amino/run/common/TestUtils.java 
core/src/test/java/amino/run/oms/KernelServerManagerTest.java

This PR suppose to get split in two one with metric aggregation and other with decision making module

VenuReddy2103 · 2019-05-06T12:22:47Z

+        /* calculate weighted moving average
+            WMA= [nDlatest+(n-1)Dsecond latest+...+Doldest]/[n+(n-1) +...+1], where n+(n-1) +...+1=n(n+1)/2
+        */
+        if (bucketSize < this.bucketSize) { // when bucket is not full


We don't have to maintain all the samples(of bucket size). We can just store either numerator of PreviousWMA(in case of weighted moving avg) OR PreviousEMA(in case of exponential moving avg) and use it for calculating the new W/EMA value.

In WMA, we don't need to calculate denominator (n*(n+1))/2 each time. Its just a summation. You can just add the new sample count to the existing one and use it. For example summation for 5 samples is 15. Summation for 6th sample is 15 + 6 = 21 i.e., also equal to (6*7)/2.

As discussed, please use exponential moving average.

Please have a look at these links below.
(https://sciencing.com/calculate-exponential-moving-averages-8221813.html)
https://www.investopedia.com/ask/answers/122314/what-exponential-moving-average-ema-formula-and-how-ema-calculated.asp
https://www.investopedia.com/ask/answers/071414/whats-difference-between-moving-average-and-weighted-moving-average.asp

Will simplify storage, fetch metrics and ema in this metrics aggregator in a separate PR. You can merge it into this one.

Raised PR #753 for it

Agreed, this whole class can be deleted.

quinton-hoole · 2019-05-10T22:01:24Z

                metric.latency = (t2 - t1) / 2;
                // TODO: This is data transfer time. Need to calculate the rate
-                metric.rate = (t3 - t2) - (t2 - t1);
+                metric.rate = ((t3 - t2) - (t2 - t1));


Don't add these unnecessary brackets.

PR #742 addresses this comment. Currently metric rate is calculate as

metric.rate = ((RANDOM_DATA_SIZE * 8 * 1000.0 * 1000.0) / (((t3 - t2) - (t2 - t1)) * 1024.0 * 1024.0));

For code clarity we are maintaining these brackets.

quinton-hoole · 2019-05-10T22:10:47Z

                    }
                };
        serverManager.registerKernelServer(info, onServerRemove);
+        metricAggregator.pushCPUMetric(info.getHost(), info.cpu);


This looks confusing. I ask myself where is OMS server pushing these metrics to? Maybe I will find out later.

MetricAggregator class has been removed and now metric are managed Node metric are managed by KernelServerManager in OMS. updateMetric is used for updating Processor count, latency and transfer rate metric.

quinton-hoole · 2019-05-10T22:29:23Z

+    // private PriorityQueue<ReplicaStat> replicaStatsHeap;
+    private MigrationPredictor predictor;
+    private transient volatile ResettableTimer timer;
+    public static int METRIC_WATCH_FREQUENCY = 30000; // in milli seconds


Need to javadoc these two constants, and what they mean.

Added javadoc and updated METRIC_WATCH_FREQUENCY variable name to MIGRATION_EVALUATION_INTERVAL.

quinton-hoole · 2019-05-10T22:30:51Z

+    public MetricWatcher(Aggregator aggregator, MicroServiceManager microServiceManager) {
+        this.aggregator = aggregator;
+        this.microServiceManager = microServiceManager;
+        this.predictor = new MigrationPredictor(aggregator, microServiceManager);


Please initialize timer in the constructor too. Otherwise it's uninitialized after the constructor has been called, which his bad.

Moved timer initialisation code from start method to constructor as per comment. Changes are taken care in PR #793.

quinton-hoole · 2019-05-10T22:35:29Z

+                        new TimerTask() {
+                            public void run() {
+                                logger.info("metric watch triggered");
+                                TreeSet<ReplicaStat> replicaStatsHeap =


Surely it's much more efficient (and simpler) to maintain the heap constantly over time? Rather than build a new heap every 30 seconds, only to throw it away and rebuild again?

Current approach for recreating heap is considered because

Replica of micro service are disposable resource in Amino.Run. So if we maintain one heap and update metric as and when migration decision triggered, Decision module will have to maintain pruning thread to remove disposed replicas.

Also TreeSet/PriorityQueue heap implementation do not have update functionality.

But still constructing TreeSet instance every 30 sec was not good approach. Now Only one instance of TreeSet is maintained. Every 30 sec TreeSet get populated and cleared after getting sorted values.

quinton-hoole · 2019-05-10T22:38:58Z

+                                        logger.warning(
+                                                String.format(
+                                                        "skipping Replica [%s] evaluation as micro service not available"));
+                                        e.printStackTrace();


This is messy, but OK. Under what conditions is the microservice not found? Surely it should always be found?

quinton-hoole · 2019-05-11T00:43:01Z

+                        String.format(
+                                "skipping kernel server [%s] as time optimization is negative",
+                                futureKS));
+                return 0;


Wait. Isn't the logic here wrong? The loop is over all inbound clients to the microservice. In this code you're not going to migrate if any of the clients will see a slowdown? The logic needs to be that the aggregate of all the clients needs to improve. That aggregate could either be the sum, or the average, or the median, or whatever.

Right?

By this check , We are ensuring that because of migration to specific kernel server if any application faces negative optimisation (processing time degraded), we ignore migrating to that kernel server.

quinton-hoole · 2019-05-11T00:47:56Z

         * @throws MicroServiceNotFoundException
         */
-        protected void pin(ServerPolicy server, InetSocketAddress host)
+        public void pin(ServerPolicy server, InetSocketAddress host)


I don't really like making this public, but OK. It would be good to find a way to prevent everything else from calling this.

Reverted above code changes. Used alternate approach for calling pin in migration.

Current approach:

Get group policy Event handler from MicroServiceManager.

Using event handler, call onNotification method and pass ReplicaID and Kernel server info as argument (Added new class MigrationNotification).

Added code in group policy for handling MigrationNotification and called pin for server policy stub (Using ReplicaID) maintained in group policy.

quinton-hoole · 2019-05-11T00:49:10Z

        serverPolicy.setReplicaId(replicaId);

+        // TODO added this code for checking migration when only one DM is set
+        serverPolicyStub.setIsLastPolicy(true);


Yes, we need to remove this please. It's not clear to me why it's needed. Can't you just test with a single DM and get the same effect?

This code was added to ensure server policy stub also maintains valid value for last policy flag. The same stub is later used for calling pin during migration where it checks for last policy.
Current approach:

Get group policy Event handler from MicroServiceManager.

Using event handler, call onNotification method and pass ReplicaID and Kernel server info as argument (Added new class MigrationNotification).

Added code in group policy for handling MigrationNotification and called pin for server policy stub (Using ReplicaID) maintained in group policy.

quinton-hoole · 2019-05-11T00:50:40Z

-    args = [project.property('omsIpFlag'), project.property('omsIp'), project.property('omsPortFlag'), project.property('omsPort')]
+    args = [project.property('omsIpFlag'), project.property('omsIp'), project.property('omsPortFlag'), project.property('omsPort'),
+            project.property('kernelServerIpFlag'), project.property('kernelServerIp'), project.property('kernelServerPortFlag'),
+            project.property('kernelServerPort4'), project.property('kernelServerLabelFlag'), project.property('kernelServerLabel4')]


This looks like a mess. What's going on here?

This code is added to start Kernel server along side Application client. In gradle task we are passing arguments required for kernel server and application to start.

quinton-hoole · 2019-05-11T00:51:26Z

+}
+
+// Customize runapp for this example
+task runappwithgetname(type: JavaExec) {


Yes, lets fix this properly. Please don't merge this code.

quinton-hoole · 2019-05-11T00:53:15Z

            // TODO: Currently for collecting metric on logging server sleep is introduced.
            //  When metric server will be available new application will get added to demonstrate metric collection
-            Thread.sleep(20000);
+            Thread.sleep(200000);


Definitely remove this. Why do you need it at all?

Removed as per comment. But addressed in PR #793

quinton-hoole · 2019-05-11T00:54:03Z

+            Thread.sleep(200000);
        } finally {
            if (oid != null) {
+                // TODO currently removing micro service delete for Migration testing


What does this comment mean? It doesn't seem to make sense.

For testing the migration, I have commented microservice deletion code. Added this comment to revert it back. In PR #793, I reverted back the commented code.

quinton-hoole · 2019-05-11T00:54:21Z


-    private static Registry getRegistry(String omsIp, int omsPort) throws Exception {
-        new KernelServerImpl(new InetSocketAddress("127.0.0.2", 11111), new InetSocketAddress(omsIp,omsPort));
+    private static Registry getRegistry(String[] args, String omsIp, int omsPort) throws Exception {


Great, thanks!

quinton-hoole

See comments. Nice job, mostly.

AmitRoushan · 2019-05-14T11:31:34Z

As this PR had commits from multiple branches to integrate with decision making module, Some of comment are applicable for those PRs. I will be ensuring above mentioned comments against PR #742 #746 #753.
Also I am maintaining commits for Decision making module only in this PR.

VenuReddy2103 · 2019-05-15T12:30:11Z

 import amino.run.kernel.common.ServerInfo;
 import amino.run.kernel.server.KernelServer;
 import amino.run.kernel.server.KernelServerImpl;
+import amino.run.oms.migrationDecision.MetricWatcher;


General convention for package name is to use lowercase
https://www.oracle.com/technetwork/java/codeconventions-135099.html

Addressed in #793 PR

VenuReddy2103 · 2019-05-16T07:08:21Z

+    private MicroServiceManager microServiceManager;
+    private KernelServerManager serverManager;
+    private MigrationPredictor predictor;
+    private transient volatile ResettableTimer timer;


Why transient volatile ?

Addressed in #793 PR

VenuReddy2103 · 2019-05-16T09:26:40Z

+
+    /** Starts metric watcher */
+    public void start() {
+        if (timer == null) {


Instead if timer != null, then simply return can be more clear ?

Addressed in #793 PR

VenuReddy2103 · 2019-05-16T09:55:13Z

+            averageExecutionTime += replicaMetric.getValue().processTime;
+
+            data = replicaMetric.getValue().dataSize;
+            nodeMetric = replicaNodeMetric.get(replicaMetric.getKey());


What we need was nodeMetric from client to the kernelserver. But we took from kernelserver to the client. Did we assume both to be same ? They both are not necessary to be same.

Addressed in #793 PR

AmitRoushan · 2019-06-04T06:35:10Z

Closing this PR as PR's remote branch is no longer linked with forked repository after Amino.Run repository moved to public.
A separate PR raised from new forked repo #793

AmitRoushan self-assigned this May 4, 2019

AmitRoushan added BeforeBarcelonaKubeCon Must be resolved before release (4/30) importance/high Degree of importance, not urgency urgency/high Degree of urgency, not importance labels May 4, 2019

AmitRoushan added this to the RC8 milestone May 4, 2019

AmitRoushan force-pushed the metricCollector branch 3 times, most recently from 529d286 to 99ece95 Compare May 6, 2019 10:13

VenuReddy2103 reviewed May 6, 2019

View reviewed changes

AmitRoushan force-pushed the metricCollector branch 6 times, most recently from 6ed63f6 to 7313c60 Compare May 8, 2019 11:31