chore(#3207): Remove pipeline element recommender (#3210)

* chore(#3207): Remove pipeline element recommender

* Merge model

* Fix layout

streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementRecommendation.java

@@ -23,8 +23,6 @@ public class PipelineElementRecommendation {
   private String elementId;
   private String name;
   private String description;
-  private Float weight;
-  private Integer count;
 
   public PipelineElementRecommendation() {
 
@@ -60,19 +58,4 @@ public void setDescription(String description) {
     this.description = description;
   }
 
-  public Float getWeight() {
-    return weight;
-  }
-
-  public void setWeight(Float weight) {
-    this.weight = weight;
-  }
-
-  public Integer getCount() {
-    return count;
-  }
-
-  public void setCount(Integer count) {
-    this.count = count;
-  }
 }

streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineElementRecommendationMessage.java

@@ -27,13 +27,11 @@
 public class PipelineElementRecommendationMessage {
 
   private List<PipelineElementRecommendation> possibleElements;
-  private List<PipelineElementRecommendation> recommendedElements;
 
   private boolean success;
 
   public PipelineElementRecommendationMessage() {
     this.possibleElements = new ArrayList<>();
-    this.recommendedElements = new ArrayList<>();
     this.success = true;
   }
 
@@ -56,15 +54,4 @@ public boolean isSuccess() {
   public void setSuccess(boolean success) {
     this.success = success;
   }
-
-  public List<PipelineElementRecommendation> getRecommendedElements() {
-    return recommendedElements;
-  }
-
-  public void setRecommendedElements(
-      List<PipelineElementRecommendation> recommendedElements) {
-    this.recommendedElements = recommendedElements;
-  }
-
-
 }

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/recommender/ElementRecommender.java

@@ -23,7 +23,6 @@
 import org.apache.streampipes.manager.matching.v2.StreamMatch;
 import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.base.NamedStreamPipesEntity;
 import org.apache.streampipes.model.graph.DataProcessorDescription;
 import org.apache.streampipes.model.graph.DataSinkDescription;
@@ -33,7 +32,6 @@
 import org.apache.streampipes.model.pipeline.PipelineElementRecommendationMessage;
 import org.apache.streampipes.model.pipeline.PipelineModification;
 import org.apache.streampipes.resource.management.SpResourceManager;
-import org.apache.streampipes.storage.api.INoSqlStorage;
 import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 
@@ -62,81 +60,13 @@ public ElementRecommender(Pipeline partialPipeline,
 
   public PipelineElementRecommendationMessage findRecommendedElements() throws NoSuitableSepasAvailableException {
     AllElementsProvider elementsProvider = new AllElementsProvider(this.pipeline);
-
-    String rootNodeId;
     try {
-      rootNodeId = getRootNodeId(elementsProvider);
       Optional<SpDataStream> outputStream = getOutputStream(elementsProvider);
       outputStream.ifPresent(spDataStream -> validate(spDataStream, getAll()));
     } catch (Exception e) {
       LOG.warn("Could not find root node or output stream of provided pipeline");
-      return recommendationMessage;
-    }
-
-    if (recommendationMessage.getPossibleElements().isEmpty()) {
-      throw new NoSuitableSepasAvailableException();
-    } else {
-      recommendationMessage
-          .setRecommendedElements(calculateWeights(
-              filterOldElements(getNoSqlStorage()
-                  .getConnectionStorageApi()
-                  .getRecommendedElements(rootNodeId))));
-      return recommendationMessage;
     }
-  }
-
-  private String getRootNodeId(AllElementsProvider elementsProvider) {
-    NamedStreamPipesEntity pe = elementsProvider.findElement(this.baseRecDomId);
-    return pe instanceof InvocableStreamPipesEntity ? ((InvocableStreamPipesEntity) pe).getBelongsTo() :
-        pe.getElementId();
-  }
-
-  private List<PipelineElementRecommendation> filterOldElements(
-      List<PipelineElementRecommendation> recommendedElements) {
-    return recommendedElements
-        .stream()
-        .filter(r -> getAll()
-            .stream()
-            .anyMatch(a -> a.getElementId().equals(r.getElementId())))
-        .collect(Collectors.toList());
-  }
-
-  private List<PipelineElementRecommendation> calculateWeights(
-      List<PipelineElementRecommendation> recommendedElements) {
-    int allConnectionsCount = recommendedElements
-        .stream()
-        .mapToInt(PipelineElementRecommendation::getCount)
-        .sum();
-
-    recommendedElements
-        .forEach(r -> {
-          r.setWeight(getWeight(r.getCount(), allConnectionsCount));
-          r.setName(getName(r.getElementId()));
-          r.setDescription(getDescription(r.getElementId()));
-        });
-
-    return recommendedElements;
-  }
-
-  private String getName(String elementId) {
-    return filter(elementId).getName();
-  }
-
-  private String getDescription(String elementId) {
-    return filter(elementId).getDescription();
-  }
-
-  private NamedStreamPipesEntity filter(String elementId) {
-    List<ConsumableStreamPipesEntity> allElements = getAll();
-    return allElements
-        .stream()
-        .filter(a -> a.getElementId().equals(elementId))
-        .findFirst()
-        .get();
-  }
-
-  private Float getWeight(Integer count, Integer allConnectionsCount) {
-    return ((float) (count)) / allConnectionsCount;
+    return recommendationMessage;
   }
 
   private void validate(SpDataStream offer, List<ConsumableStreamPipesEntity> entities) {
@@ -157,7 +87,7 @@ private void addPossibleElements(NamedStreamPipesEntity sepa) {
 
   private List<ConsumableStreamPipesEntity> getAllDataProcessors() {
     List<String> userObjects = new SpResourceManager().manageDataProcessors().findAllIdsOnly();
-    return getTripleStore()
+    return getNoSqlStore()
         .getAllDataProcessors()
         .stream()
         .filter(e -> userObjects.stream().anyMatch(u -> u.equals(e.getAppId())))
@@ -168,7 +98,7 @@ private List<ConsumableStreamPipesEntity> getAllDataProcessors() {
 
   private List<ConsumableStreamPipesEntity> getAllDataSinks() {
     List<String> userObjects = new SpResourceManager().manageDataSinks().findAllIdsOnly();
-    return getTripleStore()
+    return getNoSqlStore()
         .getAllDataSinks()
         .stream()
         .filter(e -> userObjects.stream().anyMatch(u -> u.equals(e.getAppId())))
@@ -183,14 +113,10 @@ private List<ConsumableStreamPipesEntity> getAll() {
     return allElements;
   }
 
-  private IPipelineElementDescriptionStorage getTripleStore() {
+  private IPipelineElementDescriptionStorage getNoSqlStore() {
     return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineElementDescriptionStorage();
   }
 
-  private INoSqlStorage getNoSqlStorage() {
-    return StorageDispatcher.INSTANCE.getNoSqlStore();
-  }
-
   private Optional<SpDataStream> getOutputStream(AllElementsProvider elementsProvider) {
 
     NamedStreamPipesEntity entity = elementsProvider.findElement(this.baseRecDomId);

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java

@@ -60,7 +60,6 @@ private void createDatabases() {
       // Set up streampipes internal databases
       Utils.getCouchDbUserClient();
       Utils.getCouchDbPipelineClient();
-      Utils.getCouchDbConnectionClient();
       Utils.getCouchDbNotificationClient();
       Utils.getCouchDbPipelineCategoriesClient();
 
@@ -72,7 +71,6 @@ private void createDatabases() {
 
   private void createViews() {
     addUserView();
-    addConnectionView();
     addNotificationView();
     addPipelineView();
   }
@@ -166,30 +164,4 @@ private void addUserView() {
       logFailure(PREPARING_USERS_TEXT, e);
     }
   }
-
-  private void addConnectionView() {
-    try {
-      DesignDocument connectionDocument = prepareDocument("_design/connection");
-      Map<String, MapReduce> views = new HashMap<>();
-
-      MapReduce frequentFunction = new MapReduce();
-      frequentFunction.setMap("function(doc) { if(doc.from && doc.to) { emit([doc.from, doc.to] , 1 ); } }");
-      frequentFunction.setReduce("function (key, values) { return sum(values); }");
-
-      views.put("frequent", frequentFunction);
-
-      connectionDocument.setViews(views);
-      Response resp = Utils.getCouchDbConnectionClient().design().synchronizeWithDb(connectionDocument);
-
-      if (resp.getError() != null) {
-        logFailure("Preparing database 'connection'...");
-      } else {
-        logSuccess("Preparing database 'connection'...");
-      }
-    } catch (Exception e) {
-      logFailure("Preparing database 'connection'...", e);
-    }
-  }
-
-
 }

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java

@@ -20,7 +20,6 @@
 
 import org.apache.streampipes.manager.data.PipelineGraph;
 import org.apache.streampipes.manager.data.PipelineGraphBuilder;
-import org.apache.streampipes.manager.matching.ConnectionStorageHandler;
 import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.graph.DataProcessorInvocation;
 import org.apache.streampipes.model.graph.DataSinkInvocation;
@@ -33,7 +32,7 @@
 
 public class PipelineStorageService {
 
-  private Pipeline pipeline;
+  private final Pipeline pipeline;
 
   public PipelineStorageService(Pipeline pipeline) {
     this.pipeline = pipeline;
@@ -63,8 +62,6 @@ private void preparePipeline() {
 
     pipeline.setSepas(sepas);
     pipeline.setActions(secs);
-
-    new ConnectionStorageHandler(pipeline).storeConnections();
   }
 
   private void encryptSecrets(List<InvocableStreamPipesEntity> graphs) {

streampipes-rest/src/main/java/org/apache/streampipes/rest/security/SpPermissionEvaluator.java

@@ -51,7 +51,6 @@ public boolean hasPermission(Authentication auth, Object o, Object permission) {
 
   private boolean filterRecommendation(Authentication auth, PipelineElementRecommendationMessage message) {
     Predicate<PipelineElementRecommendation> isForbidden = r -> !hasPermission(auth, r.getElementId());
-    message.getRecommendedElements().removeIf(isForbidden);
     message.getPossibleElements().removeIf(isForbidden);
 
     return true;

streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/INoSqlStorage.java

@@ -42,8 +42,6 @@ public interface INoSqlStorage {
 
   IPipelineStorage getPipelineStorageAPI();
 
-  IPipelineElementConnectionStorage getConnectionStorageApi();
-
   IUserStorage getUserStorageAPI();
 
   INotificationStorage getNotificationStorageApi();