Merge pull request #14 from TAK-Product-Center/upstream/4.7-RELEASE-32

TAK Server 4.7-RELEASE-32

README.md

@@ -4,14 +4,15 @@
 * Linux / MacOS is recommended for development. If using Windows, replace "gradlew" with "gradlew.bat" in commands below.
 
 Links:
- * [CI Test Execution](src/takserver-takcl-core/docs/ci_testing.md)
+ * [Test Execution](src/takserver-takcl-core/docs/testing.md)
+ * [Test Architecture and Development](src/takserver-takcl-core/docs/Development.md)
  * [Publishing](src/docs/publishing.md)
 
 ---
-Clean and Build TAK Server
+Clean and Build TAK Server, including war, retention service, plugin manager, user manager and schema manager.
 ```
 cd src
-./gradlew clean bootWar
+./gradlew clean bootWar bootJar shadowJar
 ```
 
 In Eclipse, choose File -> Import -> Gradle -> Existing Gradle Project
@@ -20,24 +21,24 @@ Navigate to `takserver/src`
 
 Select Finish. The TAK Server parent project, and all subprojects, will be imported into Eclipse.
 
-Install Postgres Server locally. Make sure that the PostGIS extension is also installed.
+Install PostgreSQL + PostGIS extension locally on your workstation, or run the docker container as described below. If installing locally, use 
 
 Start the Postres server.
 
-The easiest way to set up and start a Postgres server with the PostGIS plugins is to use the official PostGIS database docker container as follows and change the environment variables supplied to the container as necessary. Note the '--rm' means the container will be destroyed when it is stopped!
+To run a local PostgreSQL + PostGIS container, follow the commands below using the official PostGIS database docker container as follows, and changing the environment variables supplied to the container as necessary. Note the '--rm' means the container will be destroyed when it is stopped.
 
 ```
 docker run -it -d --rm --name TakserverServer0DB \
     --env POSTGRES_PASSWORD=e815f795745e \
     --env POSTGRES_HOST_AUTH_METHOD=trust \
     --env POSTGRES_USER=martiuser \
     --env POSTGRES_DB=cot \
-    -p 5432
+    -p 5432 postgis/postgis:10-3.1
 
 echo SQL SERVER IP: `docker inspect --format='{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' TakserverServer0DB`
 ```
 
-Setup Local Database
+Setup Local Database. If the postgis container was used, only the last two lines should be necessary.  
 ```
 - cd src/takserver-schemamanager
 - psql -d postgres  -c "CREATE ROLE martiuser LOGIN ENCRYPTED PASSWORD 'md564d5850dcafc6b4ddd03040ad1260bc2' SUPERUSER INHERIT CREATEDB NOCREATEROLE;"
@@ -51,6 +52,7 @@ Configure Local CoreConfig and Certs
 
 This is the CoreConfig that takserver war will look for when running from the takserver-core/example directory. From this point, just follow the instructions at takserver/src/docs/TAK_Server_Configuration_Guide.pdf to set up the CoreConfig and Certs. Make sure that the CoreConfig now points to the directory where the certs were generated locally.
 
+See appendix B in src/docs/TAK_Server_Configuration_Guide.pdf for cert generation instructions.
 
 ### Build and run TAK server locally for development
 
@@ -63,14 +65,14 @@ export JDK_JAVA_OPTIONS="-Dloader.path=WEB-INF/lib-provided,WEB-INF/lib,WEB-INF/
 
 TAK server consists of two processes: Messaging and API. The messaging process can run independently, but the API process needs to connect to the ignite server that runs as a part of the messaging process. For both processes, -Xmx should always be specified.
 
-Run Messaging
+Run Messaging (note - this command and the following one to run api include the **duplicatelogs** profile. This turns off the filter that blocks duplicated log messages that cause log spam in operational deployments of TAK Server.
 ```
-java -Xmx<value> -Dspring.profiles.active=messaging -jar ../build/libs/takserver-core-xyz.war
+java -Xmx<value> -Dspring.profiles.active=messaging,duplicatelogs -jar ../build/libs/takserver-core-xyz.war
 ```
 
 Run API
 ```
-java -Xmx<value> -Dspring.profiles.active=api -jar ../build/libs/takserver-core-xyz.war
+java -Xmx<value> -Dspring.profiles.active=api,duplicatelogs -jar ../build/libs/takserver-core-xyz.war
 ```
 
 Run Plugin Manager (useful when working on plugin capability)
@@ -141,7 +143,7 @@ https://localhost:8443/swagger-ui.html
 ### Integration Tests
 
 Integration tests are executed against master nightly. In addition to this, they can be executed on any branch as follows:  
-1.  Navigate to the [TAKServer Dashboard](https://git.takmaps.com/core/takserver).  
+1.  Navigate to the [TAKServer Dashboard](https://git.tak.gov/core/takserver).  
 2.  On the sidebar, hover over 'CI/CD' and select 'Pipelines'.  
 3.  Find your commit from the list and tap the Play button to the right, and select the test suite you would like to execute.  The Main suites are what is executed nightly and execute all the tests.  
 

src/federation-common/build.gradle

@@ -11,7 +11,12 @@ dependencies {
   compile project(':takserver-fig-core')
 
   // Apache Ignite (cache and distributed service grid).
-  compile group: 'org.apache.ignite', name: 'ignite-spring', version: ignite_spring_version
+//  compile group: 'org.apache.ignite', name: 'ignite-spring', version: ignite_spring_version
+//  compile group: 'org.apache.ignite', name: 'ignite-spring-cache-ext', version: ignite_spring_cache_version
+  compile group: 'org.springframework', name: 'spring-beans', version: spring_version
+  compile group: 'org.springframework', name: 'spring-context', version: spring_version
+
+
   compile group: 'org.apache.ignite', name: 'ignite-kubernetes', version: ignite_version
   compile group: 'org.apache.ignite', name: 'ignite-slf4j', version: ignite_version
 }

src/federation-common/src/main/java/tak/server/federation/FederateOutgoing.java

@@ -0,0 +1,15 @@
+package tak.server.federation;
+
+import java.util.HashSet;
+
+public class FederateOutgoing extends FederationNode {
+
+    public FederateOutgoing(FederateIdentity federateIdentity) {
+        super(federateIdentity);
+    }
+
+    public FederateOutgoing(String nodeName, FederateIdentity federateIdentity) {
+        super(nodeName, federateIdentity);
+    }
+
+}

src/federation-common/src/main/java/tak/server/federation/GuardedStreamHolder.java

@@ -2,20 +2,33 @@
 
 import static java.util.Objects.requireNonNull;
 
+import java.math.BigInteger;
+import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentSkipListSet;
 
+import javax.net.ssl.SSLSession;
+
+import org.apache.commons.lang3.RandomStringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.atakmap.Tak.BinaryBlob;
 import com.atakmap.Tak.ClientHealth;
+import com.atakmap.Tak.FederateGroups;
+import com.atakmap.Tak.FederateProvenance;
+import com.atakmap.Tak.FederatedEvent;
+import com.atakmap.Tak.ROL;
 import com.atakmap.Tak.Subscription;
 import com.google.common.base.Strings;
 
+import io.grpc.ClientCall;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
+import tak.server.federation.hub.FederationHubDependencyInjectionProxy;
 
 /*
  *
@@ -26,25 +39,50 @@ public class GuardedStreamHolder<T> {
 
     private static final Logger logger = LoggerFactory.getLogger(GuardedStreamHolder.class);
 
-    private final StreamObserver<T> clientStream;
+    private StreamObserver<T> clientStream;
+    private ClientCall<T, Subscription> clientCall;
     private long lastHealthTime;
     private ClientHealth lastHealthStatus;
-    private final FederateIdentity federateIdentity;
-    private final Subscription subscription;
+    private FederateIdentity federateIdentity;
+    private Subscription subscription;
+
+    private boolean isHub = false;
 
     private final Set<T> cache;
 
     public Set<T> getCache() {
         return cache;
     }
+
+    // for outgoing connections
+    public GuardedStreamHolder(ClientCall<T, Subscription> clientCall, String fedId, Comparator<T> comp, boolean isHub) {
+
+        requireNonNull(clientCall, "FederatedEvent groupCall");
+
+        requireNonNull(comp, "comparator");
+
+        this.isHub = isHub;
+
+        this.cache = new ConcurrentSkipListSet<T>(comp);
+
+        this.federateIdentity = new FederateIdentity(fedId);
+
+        this.clientCall = clientCall;
+
+        lastHealthTime = System.currentTimeMillis();
+        lastHealthStatus = ClientHealth.newBuilder().setStatus(ClientHealth.ServingStatus.SERVING).build();
+    }
 
-    public GuardedStreamHolder(StreamObserver<T> clientStream, String clientName, String certHash, Subscription subscription, Comparator<T> comp) {
+    // for incoming connections
+    public GuardedStreamHolder(StreamObserver<T> clientStream, String clientName, String certHash, SSLSession session, Subscription subscription, Comparator<T> comp, boolean isHub) {
 
         requireNonNull(clientStream, "FederatedEvent client stream");
 
         requireNonNull(subscription, "client subscription");
         requireNonNull(comp, "comparator");
-
+
+        this.isHub = isHub;
+
         this.cache = new ConcurrentSkipListSet<T>(comp);
 
         if (Strings.isNullOrEmpty(clientName)) {
@@ -55,7 +93,8 @@ public GuardedStreamHolder(StreamObserver<T> clientStream, String clientName, St
             throw new IllegalArgumentException("empty cert hash - invalid stream");
         }
 
-        String fedId = clientName + "-" + certHash;
+        // append a random id to the end, to prevent collisions. this is done for outgoing connections as well in the javascript code
+        String fedId = clientName + "-" + certHash  + "-" + new BigInteger(session.getId());
 
         this.subscription = subscription;
 
@@ -65,6 +104,10 @@ public GuardedStreamHolder(StreamObserver<T> clientStream, String clientName, St
         lastHealthTime = System.currentTimeMillis();
         lastHealthStatus = ClientHealth.newBuilder().setStatus(ClientHealth.ServingStatus.SERVING).build();
     }
+
+    public void setSubscription(Subscription sub) {
+    	this.subscription = sub;
+    }
 
     public void updateClientHealth(ClientHealth healthCheck) {
         this.lastHealthTime = System.currentTimeMillis();
@@ -90,17 +133,74 @@ public synchronized void send(T event) {
         if (event == null) {
             return;
         }
-
-        clientStream.onNext(event);
+
+        if (isHub) {
+        	// since hub outgoing connections can forward traffic to other hubs, we need to keep a list of visited nodes
+            // so that we can stop cycles
+            FederateProvenance prov = FederateProvenance.newBuilder()
+        			.setFederationServerId(FederationHubDependencyInjectionProxy.getInstance().fedHubServerConfig().getFullId())
+        			.setFederationServerName(FederationHubDependencyInjectionProxy.getInstance().fedHubServerConfig().getServerName())
+        			.build();
+
+            if (event instanceof FederatedEvent) {
+            	FederatedEvent fedEvent = (FederatedEvent) event;
+            	List<FederateProvenance> federateProvenances = new ArrayList<>(fedEvent.getFederateProvenanceList());
+            	federateProvenances.add(prov);
+
+            	event = (T) fedEvent.toBuilder().addAllFederateProvenance(federateProvenances).build();
+            }
+
+            if (event instanceof FederateGroups) {
+            	FederateGroups fedGroup = (FederateGroups) event;
+            	List<FederateProvenance> federateProvenances = new ArrayList<>(fedGroup.getFederateProvenanceList());
+            	federateProvenances.add(prov);
+
+            	event = (T) fedGroup.toBuilder().addAllFederateProvenance(federateProvenances).build();
+            }
+
+            if (event instanceof ROL) {
+            	ROL rol = (ROL) event;
+            	List<FederateProvenance> federateProvenances = new ArrayList<>(rol.getFederateProvenanceList());
+            	federateProvenances.add(prov);
+
+            	event = (T) rol.toBuilder().addAllFederateProvenance(federateProvenances).build();
+            }
+
+            if (event instanceof BinaryBlob) {
+            	BinaryBlob blob = (BinaryBlob) event;
+            	List<FederateProvenance> federateProvenances = new ArrayList<>(blob.getFederateProvenanceList());
+            	federateProvenances.add(prov);
+
+            	event = (T) blob.toBuilder().addAllFederateProvenance(federateProvenances).build();
+            }
+        }
+
+        // clientStream = stream of messages going from server to a connected outgoing client
+        if (clientStream != null) 
+        	clientStream.onNext(event);
+
+        // clientCall = stream of messages going from outgoing client to a server
+        if (clientCall != null)  
+        	clientCall.sendMessage(event);
     }
 
     public void throwDeadlineExceptionToClient() {
         try {
-            clientStream.onError(new StatusRuntimeException(Status.DEADLINE_EXCEEDED));
+        	if (clientStream != null)
+        		clientStream.onError(new StatusRuntimeException(Status.DEADLINE_EXCEEDED));
         } catch (Exception e) {
             logger.warn("exception sending StatusRuntimeException - DEADLINE_EXCEEDED to client", e);
         }
     }
+
+    public void throwPermissionDeniedToClient() {
+        try {
+        	if (clientStream != null)
+        		clientStream.onError(new StatusRuntimeException(Status.PERMISSION_DENIED));
+        } catch (Exception e) {
+            logger.warn("exception sending StatusRuntimeException - PERMISSION_DENIED to client", e);
+        }
+    }
 
     public FederateIdentity getFederateIdentity() {
         return federateIdentity;
@@ -111,19 +211,9 @@ public Subscription getSubscription() {
     }
 
     @Override
-    public String toString() {
-        StringBuilder builder = new StringBuilder();
-        builder.append("GuardedStreamHolder [clientStream=");
-        builder.append(clientStream);
-        builder.append(", timeSinceLastHealthCheck=");
-        builder.append(lastHealthTime);
-        builder.append(", lastHealthStatus=");
-        builder.append(lastHealthStatus);
-        builder.append(", federateIdentity=");
-        builder.append(federateIdentity);
-        builder.append(", subscription=");
-        builder.append(subscription);
-        builder.append("]");
-        return builder.toString();
-    }
+	public String toString() {
+		return "GuardedStreamHolder [clientStream=" + clientStream + ", clientCall=" + clientCall + ", lastHealthTime="
+				+ lastHealthTime + ", lastHealthStatus=" + lastHealthStatus + ", federateIdentity=" + federateIdentity
+				+ ", subscription=" + subscription + ", cache=" + cache + "]";
+	}
 }

src/federation-common/src/main/java/tak/server/federation/hub/FederationHubDependencyInjectionProxy.java

@@ -1,9 +1,11 @@
 package tak.server.federation.hub;
 
 import tak.server.federation.hub.policy.FederationHubPolicyManager;
-import tak.server.federation.hub.broker.RestartServerEvent;
 import tak.server.federation.hub.broker.SSLConfig;
+import tak.server.federation.hub.broker.events.RestartServerEvent;
 import tak.server.federation.hub.broker.FederationHubServerConfig;
+import tak.server.federation.hub.broker.HubConnectionStore;
+import tak.server.federation.hub.broker.FederationHubBroker;
 
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
@@ -77,7 +79,35 @@ public FederationHubServerConfig fedHubServerConfig() {
 
         return fedHubServerConfig;
     }
+
+    private FederationHubBroker federationHubBroker = null;
 
+    public FederationHubBroker federationHubBroker() {
+        if (federationHubBroker == null) {
+            synchronized (this) {
+                if (federationHubBroker == null) {
+                	federationHubBroker = springContext.getBean(FederationHubBroker.class);
+                }
+            }
+        }
+
+        return federationHubBroker;
+    }
+
+    private HubConnectionStore hubConnectionStore = null;
+
+    public HubConnectionStore hubConnectionStore() {
+        if (hubConnectionStore == null) {
+            synchronized (this) {
+                if (hubConnectionStore == null) {
+                	hubConnectionStore = springContext.getBean(HubConnectionStore.class);
+                }
+            }
+        }
+
+        return hubConnectionStore;
+    }
+
     public void restartV2Server() {
         springContext.publishEvent(new RestartServerEvent(this));
     }