NIFI-6510 - Analytics framework

nifi-api/src/main/java/org/apache/nifi/controller/status/ConnectionStatus.java

@@ -16,6 +16,7 @@
  */
 package org.apache.nifi.controller.status;
 
+import org.apache.nifi.controller.status.analytics.ConnectionStatusPredictions;
 import org.apache.nifi.processor.DataUnit;
 
 /**
@@ -30,6 +31,7 @@ public class ConnectionStatus implements Cloneable {
     private String destinationId;
     private String destinationName;
     private String backPressureDataSizeThreshold;
+    private ConnectionStatusPredictions predictions;
     private long backPressureBytesThreshold;
     private long backPressureObjectThreshold;
     private int inputCount;
@@ -122,6 +124,14 @@ public void setBackPressureDataSizeThreshold(String backPressureDataSizeThreshol
         setBackPressureBytesThreshold(DataUnit.parseDataSize(backPressureDataSizeThreshold, DataUnit.B).longValue());
     }
 
+    public ConnectionStatusPredictions getPredictions() {
+        return predictions;
+    }
+
+    public void setPredictions(ConnectionStatusPredictions predictions) {
+        this.predictions = predictions;
+    }
+
     public long getBackPressureObjectThreshold() {
         return backPressureObjectThreshold;
     }
@@ -202,6 +212,11 @@ public ConnectionStatus clone() {
         clonedObj.sourceName = sourceName;
         clonedObj.destinationId = destinationId;
         clonedObj.destinationName = destinationName;
+
+        if (predictions != null) {
+            clonedObj.setPredictions(predictions.clone());
+        }
+
         clonedObj.backPressureDataSizeThreshold = backPressureDataSizeThreshold;
         clonedObj.backPressureObjectThreshold = backPressureObjectThreshold;
         clonedObj.maxQueuedBytes = maxQueuedBytes;

nifi-api/src/main/java/org/apache/nifi/controller/status/analytics/ConnectionStatusPredictions.java

@@ -0,0 +1,117 @@
+/*
+ * 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.nifi.controller.status.analytics;
+
+/**
+ */
+public class ConnectionStatusPredictions implements Cloneable {
+    private long predictionIntervalMillis;
+    private int nextPredictedQueuedCount;
+    private long nextPredictedQueuedBytes;
+    private long predictedTimeToCountBackpressureMillis;
+    private long predictedTimeToBytesBackpressureMillis;
+    private int predictedPercentCount = 0;
+    private int predictedPercentBytes = 0;
+
+    public long getPredictionIntervalMillis() {
+        return predictionIntervalMillis;
+    }
+
+    public void setPredictionIntervalMillis(long predictionIntervalMillis) {
+        this.predictionIntervalMillis = predictionIntervalMillis;
+    }
+
+    public int getNextPredictedQueuedCount() {
+        return nextPredictedQueuedCount;
+    }
+
+    public void setNextPredictedQueuedCount(int nextPredictedQueuedCount) {
+        this.nextPredictedQueuedCount = nextPredictedQueuedCount;
+    }
+
+    public long getNextPredictedQueuedBytes() {
+        return nextPredictedQueuedBytes;
+    }
+
+    public void setNextPredictedQueuedBytes(long nextPredictedQueuedBytes) {
+        this.nextPredictedQueuedBytes = nextPredictedQueuedBytes;
+    }
+
+    public long getPredictedTimeToCountBackpressureMillis() {
+        return predictedTimeToCountBackpressureMillis;
+    }
+
+    public void setPredictedTimeToCountBackpressureMillis(long predictedTimeToCountBackpressureMillis) {
+        this.predictedTimeToCountBackpressureMillis = predictedTimeToCountBackpressureMillis;
+    }
+
+    public long getPredictedTimeToBytesBackpressureMillis() {
+        return predictedTimeToBytesBackpressureMillis;
+    }
+
+    public void setPredictedTimeToBytesBackpressureMillis(long predictedTimeToBytesBackpressureMillis) {
+        this.predictedTimeToBytesBackpressureMillis = predictedTimeToBytesBackpressureMillis;
+    }
+
+    public int getPredictedPercentCount() {
+        return predictedPercentCount;
+    }
+
+    public void setPredictedPercentCount(int predictedPercentCount) {
+        this.predictedPercentCount = predictedPercentCount;
+    }
+
+    public int getPredictedPercentBytes() {
+        return predictedPercentBytes;
+    }
+
+    public void setPredictedPercentBytes(int predictedPercentBytes) {
+        this.predictedPercentBytes = predictedPercentBytes;
+    }
+
+    @Override
+    public ConnectionStatusPredictions clone() {
+        final ConnectionStatusPredictions clonedObj = new ConnectionStatusPredictions();
+        clonedObj.nextPredictedQueuedBytes = nextPredictedQueuedBytes;
+        clonedObj.nextPredictedQueuedCount = nextPredictedQueuedCount;
+        clonedObj.predictedTimeToBytesBackpressureMillis = predictedTimeToBytesBackpressureMillis;
+        clonedObj.predictedTimeToCountBackpressureMillis = predictedTimeToCountBackpressureMillis;
+        clonedObj.predictedPercentCount = predictedPercentCount;
+        clonedObj.predictedPercentBytes = predictedPercentBytes;
+        return clonedObj;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("ConnectionStatusPredictions [id=");
+        builder.append(", nextPredictedQueuedBytes=");
+        builder.append(nextPredictedQueuedBytes);
+        builder.append(", nextPredictedQueuedCount=");
+        builder.append(nextPredictedQueuedCount);
+        builder.append(", predictedTimeToBytesBackpressureMillis=");
+        builder.append(predictedTimeToBytesBackpressureMillis);
+        builder.append(", predictedTimeToCountBackpressureMillis=");
+        builder.append(predictedTimeToCountBackpressureMillis);
+        builder.append(", predictedPercentCount=");
+        builder.append(predictedPercentCount);
+        builder.append(", predictedPercentBytes=");
+        builder.append(predictedPercentBytes);
+        builder.append("]");
+        return builder.toString();
+    }
+}

nifi-api/src/main/java/org/apache/nifi/controller/status/analytics/QueryWindow.java

@@ -0,0 +1,62 @@
+/*
+ * 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.nifi.controller.status.analytics;
+
+import java.util.Date;
+
+/**
+ * QueryWindow is used to track time intervals for querying for observations
+ */
+public class QueryWindow {
+
+    private long startTimeMillis;
+    private long endTimeMillis;
+
+    public QueryWindow(long startTimeMillis, long endTimeMillis) {
+        this.startTimeMillis = startTimeMillis;
+        this.endTimeMillis = endTimeMillis;
+    }
+
+    public long getStartTimeMillis() {
+        return startTimeMillis;
+    }
+
+    public void setStartTimeMillis(long startTimeMillis) {
+        this.startTimeMillis = startTimeMillis;
+    }
+
+    public long getEndTimeMillis() {
+        return endTimeMillis;
+    }
+
+    public void setEndTimeMillis(long endTimeMillis) {
+        this.endTimeMillis = endTimeMillis;
+    }
+
+    public Date getStartDateTime() {
+        return new Date(startTimeMillis);
+    }
+
+    public Date getEndDateTime() {
+        return new Date(endTimeMillis);
+    }
+
+    public long getTimeDifferenceMillis(){
+        return endTimeMillis - startTimeMillis;
+    }
+
+}

nifi-api/src/main/java/org/apache/nifi/controller/status/analytics/StatusAnalytics.java

@@ -0,0 +1,46 @@
+/*
+ * 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.nifi.controller.status.analytics;
+
+import java.util.Map;
+
+/**
+ * The StatusAnalytics interface offers methods for accessing predicted and other values for a single component (Connection instance, e.g.).
+ */
+public interface StatusAnalytics {
+
+    /**
+     * Get the Query Window used by the analytics instance
+     * @return queryWindow
+     */
+    QueryWindow getQueryWindow();
+
+    /**
+     * Get available predictions where the key (String) in the map is the name of the prediction and value (Long)
+     * is the value for the prediction
+     * @return map
+     */
+    Map<String,Long> getPredictions();
+
+    /**
+     * Return if analytics object supports online learning
+     * @return boolean
+     */
+    boolean supportsOnlineLearning();
+
+
+}

nifi-api/src/main/java/org/apache/nifi/controller/status/analytics/StatusAnalyticsEngine.java

@@ -0,0 +1,28 @@
+/*
+ * 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.nifi.controller.status.analytics;
+
+public interface StatusAnalyticsEngine {
+
+    /**
+     * Retrieve status analytics object for given component
+     * @param componentId identifier for component
+     * @return componenet specific status analytics object
+     */
+    StatusAnalytics getStatusAnalytics(String componentId);
+
+}

nifi-api/src/main/java/org/apache/nifi/controller/status/analytics/StatusAnalyticsModel.java

@@ -0,0 +1,64 @@
+/*
+ * 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.nifi.controller.status.analytics;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+public interface StatusAnalyticsModel {
+
+    /**
+     * Train model with provided observations (features, labels/targets)
+     * @param features Stream of feature observation values
+     * @param labels target observation values
+     */
+    void learn(Stream<Double[]> features, Stream<Double> labels);
+
+    /**
+     * Return a prediction given observation values
+     * @param feature feature observation values values
+     * @return prediction of target/label
+     */
+    Double predict(Double[] feature);
+
+    /**
+     * Predict a feature given a known target and known predictor values (if multiple predictors are included with model)
+     * @param predictVariableIndex index of feature that we would like to predict (index should align with order provided in model learn method)
+     * @param knownVariablesWithIndex a map of known predictor values with their indexes if available
+     * @param label known target value
+     * @return prediction for variable
+     */
+    Double predictVariable(Integer predictVariableIndex, Map<Integer,Double> knownVariablesWithIndex, Double label);
+
+    /**
+     * Indicate if model supports online learning (e.g. can learn new observation samples to create a model)
+     * @return boolean indicating online learning support
+     */
+    Boolean supportsOnlineLearning();
+
+    /**
+     * Returns a map of scores relevant to model (e.g. rSquared, Confidence Intervals, etc.)
+     * @return Map of score names with values
+     */
+    Map<String,Double> getScores();
+
+    /**
+     * Resets a model by clearing observations and other calculations
+     */
+    void clear();
+
+}

nifi-assembly/LICENSE

@@ -2583,6 +2583,31 @@ This product bundles 'lodash' which is available under an MIT license.
     licenses; we recommend you read them, as their terms may differ from the
     terms above.
 
+This product bundles 'moment' which is available under an MIT license.
+
+    Copyright (c) JS Foundation and other contributors
+
+    Permission is hereby granted, free of charge, to any person
+    obtaining a copy of this software and associated documentation
+    files (the "Software"), to deal in the Software without
+    restriction, including without limitation the rights to use,
+    copy, modify, merge, publish, distribute, sublicense, and/or sell
+    copies of the Software, and to permit persons to whom the
+    Software is furnished to do so, subject to the following
+    conditions:
+
+    The above copyright notice and this permission notice shall be
+    included in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+    EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+    OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+    NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+    HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+    OTHER DEALINGS IN THE SOFTWARE.
+
 The binary distribution of this product bundles 'normalize.css'
 
   NORMALIZE.CSS LICENSE