datarouter-prov/src/main/java/org/onap/dmaap/datarouter/provisioning/SynchronizerTask.java

   1 /*******************************************************************************
   2  * ============LICENSE_START==================================================
   3  * * org.onap.dmaap
   4  * * ===========================================================================
   5  * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * * ===========================================================================
   7  * * Licensed under the Apache License, Version 2.0 (the "License");
   8  * * you may not use this file except in compliance with the License.
   9  * * You may obtain a copy of the License at
  10  * *
  11  *  *      http://www.apache.org/licenses/LICENSE-2.0
  12  * *
  13  *  * Unless required by applicable law or agreed to in writing, software
  14  * * distributed under the License is distributed on an "AS IS" BASIS,
  15  * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * * See the License for the specific language governing permissions and
  17  * * limitations under the License.
  18  * * ============LICENSE_END====================================================
  19  * *
  20  * * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  21  * *
  22  ******************************************************************************/
  23
  24
  25 package org.onap.dmaap.datarouter.provisioning;
  26
  27 import java.io.ByteArrayOutputStream;
  28 import java.io.File;
  29 import java.io.FileInputStream;
  30 import java.io.InputStream;
  31 import java.net.InetAddress;
  32 import java.net.UnknownHostException;
  33 import java.nio.file.Files;
  34 import java.nio.file.Path;
  35 import java.nio.file.Paths;
  36 import java.nio.file.StandardCopyOption;
  37 import java.security.KeyStore;
  38 import java.sql.Connection;
  39 import java.sql.SQLException;
  40 import java.util.ArrayList;
  41 import java.util.Arrays;
  42 import java.util.Collection;
  43 import java.util.HashMap;
  44 import java.util.Map;
  45 import java.util.Properties;
  46 import java.util.Set;
  47 import java.util.Timer;
  48 import java.util.TimerTask;
  49 import java.util.TreeSet;
  50
  51 import javax.servlet.http.HttpServletResponse;
  52
  53 import org.apache.http.HttpEntity;
  54 import org.apache.http.HttpResponse;
  55 import org.apache.http.client.methods.HttpGet;
  56 import org.apache.http.client.methods.HttpPost;
  57 import org.apache.http.conn.scheme.Scheme;
  58 import org.apache.http.conn.ssl.SSLSocketFactory;
  59 import org.apache.http.entity.ByteArrayEntity;
  60 import org.apache.http.entity.ContentType;
  61 import org.apache.http.impl.client.AbstractHttpClient;
  62 import org.apache.http.impl.client.DefaultHttpClient;
  63 import org.apache.log4j.Logger;
  64 import org.json.JSONArray;
  65 import org.json.JSONException;
  66 import org.json.JSONObject;
  67 import org.json.JSONTokener;
  68 import org.onap.dmaap.datarouter.provisioning.beans.EgressRoute;
  69 import org.onap.dmaap.datarouter.provisioning.beans.Feed;
  70 import org.onap.dmaap.datarouter.provisioning.beans.Group;
  71 import org.onap.dmaap.datarouter.provisioning.beans.IngressRoute;
  72 import org.onap.dmaap.datarouter.provisioning.beans.NetworkRoute;
  73 import org.onap.dmaap.datarouter.provisioning.beans.Parameters;
  74 import org.onap.dmaap.datarouter.provisioning.beans.Subscription;
  75 import org.onap.dmaap.datarouter.provisioning.beans.Syncable;
  76 import org.onap.dmaap.datarouter.provisioning.utils.DB;
  77 import org.onap.dmaap.datarouter.provisioning.utils.LogfileLoader;
  78 import org.onap.dmaap.datarouter.provisioning.utils.RLEBitSet;
  79 import org.onap.dmaap.datarouter.provisioning.utils.URLUtilities;
  80
  81 /**
  82  * This class handles synchronization between provisioning servers (PODs).  It has three primary functions:
  83  * <ol>
  84  * <li>Checking DNS once per minute to see which POD the DNS CNAME points to. The CNAME will point to
  85  * the active (master) POD.</li>
  86  * <li>On non-master (standby) PODs, fetches provisioning data and logs in order to keep MariaDB in sync.</li>
  87  * <li>Providing information to other parts of the system as to the current role (ACTIVE, STANDBY, UNKNOWN)
  88  * of this POD.</li>
  89  * </ol>
  90  * <p>For this to work correctly, the following code needs to be placed at the beginning of main().</p>
  91  * <code>
  92  * Security.setProperty("networkaddress.cache.ttl", "10");
  93  * </code>
  94  *
  95  * @author Robert Eby
  96  * @version $Id: SynchronizerTask.java,v 1.10 2014/03/21 13:50:10 eby Exp $
  97  */
  98 public class SynchronizerTask extends TimerTask {
  99
 100     /**
 101      * This is a singleton -- there is only one SynchronizerTask object in the server
 102      */
 103     private static SynchronizerTask synctask;
 104
 105     /**
 106      * This POD is unknown -- not on the list of PODs
 107      */
 108     public static final int UNKNOWN = 0;
 109     /**
 110      * This POD is active -- on the list of PODs, and the DNS CNAME points to us
 111      */
 112     public static final int ACTIVE = 1;
 113     /**
 114      * This POD is standby -- on the list of PODs, and the DNS CNAME does not point to us
 115      */
 116     public static final int STANDBY = 2;
 117     private static final String[] stnames = {"UNKNOWN", "ACTIVE", "STANDBY"};
 118     private static final long ONE_HOUR = 60 * 60 * 1000L;
 119
 120     private final Logger logger;
 121     private final Timer rolex;
 122     private final String spooldir;
 123     private int state;
 124     private boolean doFetch;
 125     private long nextsynctime;
 126     private AbstractHttpClient httpclient = null;
 127
 128     /**
 129      * Get the singleton SynchronizerTask object.
 130      *
 131      * @return the SynchronizerTask
 132      */
 133     public static synchronized SynchronizerTask getSynchronizer() {
 134         if (synctask == null) {
 135             synctask = new SynchronizerTask();
 136         }
 137         return synctask;
 138     }
 139
 140     @SuppressWarnings("deprecation")
 141     private SynchronizerTask() {
 142         logger = Logger.getLogger("org.onap.dmaap.datarouter.provisioning.internal");
 143         rolex = new Timer();
 144         spooldir = (new DB()).getProperties().getProperty("org.onap.dmaap.datarouter.provserver.spooldir");
 145         state = UNKNOWN;
 146         doFetch = true;        // start off with a fetch
 147         nextsynctime = 0;
 148
 149         logger.info("PROV5000: Sync task starting, server state is UNKNOWN");
 150         try {
 151             Properties props = (new DB()).getProperties();
 152             String type = props.getProperty(Main.KEYSTORE_TYPE_PROPERTY, "jks");
 153             String store = props.getProperty(Main.KEYSTORE_PATH_PROPERTY);
 154             String pass = props.getProperty(Main.KEYSTORE_PASSWORD_PROPERTY);
 155             KeyStore keyStore = KeyStore.getInstance(type);
 156             try(FileInputStream instream = new FileInputStream(new File(store))) {
 157                 keyStore.load(instream, pass.toCharArray());
 158
 159             }
 160                 store = props.getProperty(Main.TRUSTSTORE_PATH_PROPERTY);
 161                 pass = props.getProperty(Main.TRUSTSTORE_PASSWORD_PROPERTY);
 162                 KeyStore trustStore = null;
 163                 if (store != null && store.length() > 0) {
 164                     trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
 165                     try(FileInputStream instream = new FileInputStream(new File(store))){
 166                         trustStore.load(instream, pass.toCharArray());
 167
 168                     }
 169                 }
 170
 171             // We are connecting with the node name, but the certificate will have the CNAME
 172             // So we need to accept a non-matching certificate name
 173             String keystorepass = props.getProperty(
 174                 Main.KEYSTORE_PASSWORD_PROPERTY); //itrack.web.att.com/browse/DATARTR-6 for changing hard coded passphase ref
 175            try(AbstractHttpClient hc = new DefaultHttpClient()) {
 176                SSLSocketFactory socketFactory =
 177                        (trustStore == null)
 178                                ? new SSLSocketFactory(keyStore, keystorepass)
 179                                : new SSLSocketFactory(keyStore, keystorepass, trustStore);
 180                socketFactory.setHostnameVerifier(SSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
 181                Scheme sch = new Scheme("https", 443, socketFactory);
 182                hc.getConnectionManager().getSchemeRegistry().register(sch);
 183             httpclient = hc;
 184            }
 185             // Run once every 5 seconds to check DNS, etc.
 186             long interval = 0;
 187             try {
 188                 String s = props.getProperty("org.onap.dmaap.datarouter.provserver.sync_interval", "5000");
 189                 interval = Long.parseLong(s);
 190             } catch (NumberFormatException e) {
 191                 interval = 5000L;
 192             }
 193             rolex.scheduleAtFixedRate(this, 0L, interval);
 194         } catch (Exception e) {
 195             logger.warn("PROV5005: Problem starting the synchronizer: " + e);
 196         }
 197     }
 198
 199     /**
 200      * What is the state of this POD?
 201      *
 202      * @return one of ACTIVE, STANDBY, UNKNOWN
 203      */
 204     public int getState() {
 205         return state;
 206     }
 207
 208     /**
 209      * Is this the active POD?
 210      *
 211      * @return true if we are active (the master), false otherwise
 212      */
 213     public boolean isActive() {
 214         return state == ACTIVE;
 215     }
 216
 217     /**
 218      * This method is used to signal that another POD (the active POD) has sent us a /fetchProv request, and that we
 219      * should re-synchronize with the master.
 220      */
 221     public void doFetch() {
 222         doFetch = true;
 223     }
 224
 225     /**
 226      * Runs once a minute in order to <ol>
 227      * <li>lookup DNS names,</li>
 228      * <li>determine the state of this POD,</li>
 229      * <li>if this is a standby POD, and the fetch flag is set, perform a fetch of state from the active POD.</li>
 230      * <li>if this is a standby POD, check if there are any new log records to be replicated.</li>
 231      * </ol>
 232      */
 233     @Override
 234     public void run() {
 235         try {
 236             state = lookupState();
 237             if (state == STANDBY) {
 238                 // Only copy provisioning data FROM the active server TO the standby
 239                 if (doFetch || (System.currentTimeMillis() >= nextsynctime)) {
 240                     logger.debug("Initiating a sync...");
 241                     JSONObject jo = readProvisioningJSON();
 242                     if (jo != null) {
 243                         doFetch = false;
 244                         syncFeeds(jo.getJSONArray("feeds"));
 245                         syncSubs(jo.getJSONArray("subscriptions"));
 246                         syncGroups(jo.getJSONArray("groups")); //Rally:US708115 - 1610
 247                         syncParams(jo.getJSONObject("parameters"));
 248                         // The following will not be present in a version=1.0 provfeed
 249                         JSONArray ja = jo.optJSONArray("ingress");
 250                         if (ja != null) {
 251                             syncIngressRoutes(ja);
 252                         }
 253                         JSONObject j2 = jo.optJSONObject("egress");
 254                         if (j2 != null) {
 255                             syncEgressRoutes(j2);
 256                         }
 257                         ja = jo.optJSONArray("routing");
 258                         if (ja != null) {
 259                             syncNetworkRoutes(ja);
 260                         }
 261                     }
 262                     logger.info("PROV5013: Sync completed.");
 263                     nextsynctime = System.currentTimeMillis() + ONE_HOUR;
 264                 }
 265             } else {
 266                 // Don't do fetches on non-standby PODs
 267                 doFetch = false;
 268             }
 269
 270             // Fetch DR logs as needed - server to server
 271             LogfileLoader lfl = LogfileLoader.getLoader();
 272             if (lfl.isIdle()) {
 273                 // Only fetch new logs if the loader is waiting for them.
 274                 logger.trace("Checking for logs to replicate...");
 275                 RLEBitSet local = lfl.getBitSet();
 276                 RLEBitSet remote = readRemoteLoglist();
 277                 remote.andNot(local);
 278                 if (!remote.isEmpty()) {
 279                     logger.debug(" Replicating logs: " + remote);
 280                     replicateDRLogs(remote);
 281                 }
 282             }
 283         } catch (Exception e) {
 284             logger.warn("PROV0020: Caught exception in SynchronizerTask: " + e);
 285             e.printStackTrace();
 286         }
 287     }
 288
 289     /**
 290      * This method is used to lookup the CNAME that points to the active server. It returns 0 (UNKNOWN), 1(ACTIVE), or 2
 291      * (STANDBY) to indicate the state of this server.
 292      *
 293      * @return the current state
 294      */
 295     private int lookupState() {
 296         int newstate = UNKNOWN;
 297         try {
 298             InetAddress myaddr = InetAddress.getLocalHost();
 299             if (logger.isTraceEnabled()) {
 300                 logger.trace("My address: " + myaddr);
 301             }
 302             String thisPod = myaddr.getHostName();
 303             Set<String> pods = new TreeSet<>(Arrays.asList(BaseServlet.getPods()));
 304             if (pods.contains(thisPod)) {
 305                 InetAddress pserver = InetAddress.getByName(BaseServlet.activeProvName);
 306                 newstate = myaddr.equals(pserver) ? ACTIVE : STANDBY;
 307                 if (logger.isDebugEnabled() && System.currentTimeMillis() >= nextMsg) {
 308                     logger.debug("Active POD = " + pserver + ", Current state is " + stnames[newstate]);
 309                     nextMsg = System.currentTimeMillis() + (5 * 60 * 1000L);
 310                 }
 311             } else {
 312                 logger.warn("PROV5003: My name (" + thisPod + ") is missing from the list of provisioning servers.");
 313             }
 314         } catch (UnknownHostException e) {
 315             logger.warn("PROV5002: Cannot determine the name of this provisioning server.");
 316         }
 317
 318         if (newstate != state) {
 319             logger
 320                 .info(String.format("PROV5001: Server state changed from %s to %s", stnames[state], stnames[newstate]));
 321         }
 322         return newstate;
 323     }
 324
 325     private static long nextMsg = 0;    // only display the "Current state" msg every 5 mins.
 326
 327     /**
 328      * Synchronize the Feeds in the JSONArray, with the Feeds in the DB.
 329      */
 330     private void syncFeeds(JSONArray ja) {
 331         Collection<Syncable> coll = new ArrayList<Syncable>();
 332         for (int n = 0; n < ja.length(); n++) {
 333             try {
 334                 Feed f = new Feed(ja.getJSONObject(n));
 335                 coll.add(f);
 336             } catch (Exception e) {
 337                 logger.warn("PROV5004: Invalid object in feed: " + ja.optJSONObject(n));
 338             }
 339         }
 340         if (sync(coll, Feed.getAllFeeds())) {
 341             BaseServlet.provisioningDataChanged();
 342         }
 343     }
 344
 345     /**
 346      * Synchronize the Subscriptions in the JSONArray, with the Subscriptions in the DB.
 347      */
 348     private void syncSubs(JSONArray ja) {
 349         Collection<Syncable> coll = new ArrayList<Syncable>();
 350         for (int n = 0; n < ja.length(); n++) {
 351             try {
 352                 //Data Router Subscriber HTTPS Relaxation feature USERSTORYID:US674047.
 353                 JSONObject j = ja.getJSONObject(n);
 354                 j.put("sync", "true");
 355                 Subscription s = new Subscription(j);
 356                 coll.add(s);
 357             } catch (Exception e) {
 358                 logger.warn("PROV5004: Invalid object in subscription: " + ja.optJSONObject(n));
 359             }
 360         }
 361         if (sync(coll, Subscription.getAllSubscriptions())) {
 362             BaseServlet.provisioningDataChanged();
 363         }
 364     }
 365
 366     /**
 367      * Rally:US708115  - Synchronize the Groups in the JSONArray, with the Groups in the DB.
 368      */
 369     private void syncGroups(JSONArray ja) {
 370         Collection<Syncable> coll = new ArrayList<Syncable>();
 371         for (int n = 0; n < ja.length(); n++) {
 372             try {
 373                 Group g = new Group(ja.getJSONObject(n));
 374                 coll.add(g);
 375             } catch (Exception e) {
 376                 logger.warn("PROV5004: Invalid object in subscription: " + ja.optJSONObject(n));
 377             }
 378         }
 379         if (sync(coll, Group.getAllgroups())) {
 380             BaseServlet.provisioningDataChanged();
 381         }
 382     }
 383
 384
 385     /**
 386      * Synchronize the Parameters in the JSONObject, with the Parameters in the DB.
 387      */
 388     private void syncParams(JSONObject jo) {
 389         Collection<Syncable> coll = new ArrayList<Syncable>();
 390         for (String k : jo.keySet()) {
 391             String v = "";
 392             try {
 393                 v = jo.getString(k);
 394             } catch (JSONException e) {
 395                 try {
 396                     v = "" + jo.getInt(k);
 397                 } catch (JSONException e1) {
 398                     JSONArray ja = jo.getJSONArray(k);
 399                     for (int i = 0; i < ja.length(); i++) {
 400                         if (i > 0) {
 401                             v += "|";
 402                         }
 403                         v += ja.getString(i);
 404                     }
 405                 }
 406             }
 407             coll.add(new Parameters(k, v));
 408         }
 409         if (sync(coll, Parameters.getParameterCollection())) {
 410             BaseServlet.provisioningDataChanged();
 411             BaseServlet.provisioningParametersChanged();
 412         }
 413     }
 414
 415     private void syncIngressRoutes(JSONArray ja) {
 416         Collection<Syncable> coll = new ArrayList<Syncable>();
 417         for (int n = 0; n < ja.length(); n++) {
 418             try {
 419                 IngressRoute in = new IngressRoute(ja.getJSONObject(n));
 420                 coll.add(in);
 421             } catch (NumberFormatException e) {
 422                 logger.warn("PROV5004: Invalid object in ingress routes: " + ja.optJSONObject(n));
 423             }
 424         }
 425         if (sync(coll, IngressRoute.getAllIngressRoutes())) {
 426             BaseServlet.provisioningDataChanged();
 427         }
 428     }
 429
 430     private void syncEgressRoutes(JSONObject jo) {
 431         Collection<Syncable> coll = new ArrayList<Syncable>();
 432         for (String key : jo.keySet()) {
 433             try {
 434                 int sub = Integer.parseInt(key);
 435                 String node = jo.getString(key);
 436                 EgressRoute er = new EgressRoute(sub, node);
 437                 coll.add(er);
 438             } catch (NumberFormatException e) {
 439                 logger.warn("PROV5004: Invalid subid in egress routes: " + key);
 440             } catch (IllegalArgumentException e) {
 441                 logger.warn("PROV5004: Invalid node name in egress routes: " + key);
 442             }
 443         }
 444         if (sync(coll, EgressRoute.getAllEgressRoutes())) {
 445             BaseServlet.provisioningDataChanged();
 446         }
 447     }
 448
 449     private void syncNetworkRoutes(JSONArray ja) {
 450         Collection<Syncable> coll = new ArrayList<Syncable>();
 451         for (int n = 0; n < ja.length(); n++) {
 452             try {
 453                 NetworkRoute nr = new NetworkRoute(ja.getJSONObject(n));
 454                 coll.add(nr);
 455             } catch (JSONException e) {
 456                 logger.warn("PROV5004: Invalid object in network routes: " + ja.optJSONObject(n));
 457             }
 458         }
 459         if (sync(coll, NetworkRoute.getAllNetworkRoutes())) {
 460             BaseServlet.provisioningDataChanged();
 461         }
 462     }
 463
 464     private boolean sync(Collection<? extends Syncable> newc, Collection<? extends Syncable> oldc) {
 465         boolean changes = false;
 466         try {
 467             Map<String, Syncable> newmap = getMap(newc);
 468             Map<String, Syncable> oldmap = getMap(oldc);
 469             Set<String> union = new TreeSet<String>(newmap.keySet());
 470             union.addAll(oldmap.keySet());
 471             DB db = new DB();
 472             @SuppressWarnings("resource")
 473             Connection conn = db.getConnection();
 474             for (String n : union) {
 475                 Syncable newobj = newmap.get(n);
 476                 Syncable oldobj = oldmap.get(n);
 477                 if (oldobj == null) {
 478                     if (logger.isDebugEnabled()) {
 479                         logger.debug("  Inserting record: " + newobj);
 480                     }
 481                     newobj.doInsert(conn);
 482                     changes = true;
 483                 } else if (newobj == null) {
 484                     if (logger.isDebugEnabled()) {
 485                         logger.debug("  Deleting record: " + oldobj);
 486                     }
 487                     oldobj.doDelete(conn);
 488                     changes = true;
 489                 } else if (!newobj.equals(oldobj)) {
 490                     if (logger.isDebugEnabled()) {
 491                         logger.debug("  Updating record: " + newobj);
 492                     }
 493                     newobj.doUpdate(conn);
 494
 495                     /**Rally US708115
 496                      * Change Ownership of FEED - 1610, Syncronised with secondary DB.
 497                      * */
 498                     checkChnageOwner(newobj, oldobj);
 499
 500                     changes = true;
 501                 }
 502             }
 503             db.release(conn);
 504         } catch (SQLException e) {
 505             logger.warn("PROV5009: problem during sync, exception: " + e);
 506             e.printStackTrace();
 507         }
 508         return changes;
 509     }
 510
 511     private Map<String, Syncable> getMap(Collection<? extends Syncable> c) {
 512         Map<String, Syncable> map = new HashMap<String, Syncable>();
 513         for (Syncable v : c) {
 514             map.put(v.getKey(), v);
 515         }
 516         return map;
 517     }
 518
 519     /**Change owner of FEED/SUBSCRIPTION*/
 520     /**
 521      * Rally US708115 Change Ownership of FEED - 1610
 522      */
 523     private void checkChnageOwner(Syncable newobj, Syncable oldobj) {
 524         if (newobj instanceof Feed) {
 525             Feed oldfeed = (Feed) oldobj;
 526             Feed newfeed = (Feed) newobj;
 527
 528             if (!oldfeed.getPublisher().equals(newfeed.getPublisher())) {
 529                 logger.info("PROV5013 -  Previous publisher: " + oldfeed.getPublisher() + ": New publisher-" + newfeed
 530                     .getPublisher());
 531                 oldfeed.setPublisher(newfeed.getPublisher());
 532                 oldfeed.changeOwnerShip();
 533             }
 534         } else if (newobj instanceof Subscription) {
 535             Subscription oldsub = (Subscription) oldobj;
 536             Subscription newsub = (Subscription) newobj;
 537
 538             if (!oldsub.getSubscriber().equals(newsub.getSubscriber())) {
 539                 logger.info("PROV5013 -  Previous subscriber: " + oldsub.getSubscriber() + ": New subscriber-" + newsub
 540                     .getSubscriber());
 541                 oldsub.setSubscriber(newsub.getSubscriber());
 542                 oldsub.changeOwnerShip();
 543             }
 544         }
 545
 546     }
 547
 548     /**
 549      * Issue a GET on the peer POD's /internal/prov/ URL to get a copy of its provisioning data.
 550      *
 551      * @return the provisioning data (as a JONObject)
 552      */
 553     private synchronized JSONObject readProvisioningJSON() {
 554         String url = URLUtilities.generatePeerProvURL();
 555         HttpGet get = new HttpGet(url);
 556         try {
 557             HttpResponse response = httpclient.execute(get);
 558             int code = response.getStatusLine().getStatusCode();
 559             if (code != HttpServletResponse.SC_OK) {
 560                 logger.warn("PROV5010: readProvisioningJSON failed, bad error code: " + code);
 561                 return null;
 562             }
 563             HttpEntity entity = response.getEntity();
 564             String ctype = entity.getContentType().getValue().trim();
 565             if (!ctype.equals(BaseServlet.PROVFULL_CONTENT_TYPE1) && !ctype
 566                 .equals(BaseServlet.PROVFULL_CONTENT_TYPE2)) {
 567                 logger.warn("PROV5011: readProvisioningJSON failed, bad content type: " + ctype);
 568                 return null;
 569             }
 570             return new JSONObject(new JSONTokener(entity.getContent()));
 571         } catch (Exception e) {
 572             logger.warn("PROV5012: readProvisioningJSON failed, exception: " + e);
 573             return null;
 574         } finally {
 575             get.releaseConnection();
 576         }
 577     }
 578
 579     /**
 580      * Issue a GET on the peer POD's /internal/drlogs/ URL to get an RELBitSet representing the log records available in
 581      * the remote database.
 582      *
 583      * @return the bitset
 584      */
 585     private RLEBitSet readRemoteLoglist() {
 586         RLEBitSet bs = new RLEBitSet();
 587         String url = URLUtilities.generatePeerLogsURL();
 588
 589         //Fixing if only one Prov is configured, not to give exception to fill logs, return empty bitset.
 590         if (url.equals("")) {
 591             return bs;
 592         }
 593         //End of fix.
 594
 595         HttpGet get = new HttpGet(url);
 596         try {
 597             HttpResponse response = httpclient.execute(get);
 598             int code = response.getStatusLine().getStatusCode();
 599             if (code != HttpServletResponse.SC_OK) {
 600                 logger.warn("PROV5010: readRemoteLoglist failed, bad error code: " + code);
 601                 return bs;
 602             }
 603             HttpEntity entity = response.getEntity();
 604             String ctype = entity.getContentType().getValue().trim();
 605             if (!ctype.equals("text/plain")) {
 606                 logger.warn("PROV5011: readRemoteLoglist failed, bad content type: " + ctype);
 607                 return bs;
 608             }
 609             InputStream is = entity.getContent();
 610             ByteArrayOutputStream bos = new ByteArrayOutputStream();
 611             int ch = 0;
 612             while ((ch = is.read()) >= 0) {
 613                 bos.write(ch);
 614             }
 615             bs.set(bos.toString());
 616             is.close();
 617         } catch (Exception e) {
 618             logger.warn("PROV5012: readRemoteLoglist failed, exception: " + e);
 619             return bs;
 620         } finally {
 621             get.releaseConnection();
 622         }
 623         return bs;
 624     }
 625
 626     /**
 627      * Issue a POST on the peer POD's /internal/drlogs/ URL to fetch log records available in the remote database that
 628      * we wish to copy to the local database.
 629      *
 630      * @param bs the bitset (an RELBitSet) of log records to fetch
 631      */
 632     private void replicateDRLogs(RLEBitSet bs) {
 633         String url = URLUtilities.generatePeerLogsURL();
 634         HttpPost post = new HttpPost(url);
 635         try {
 636             String t = bs.toString();
 637             HttpEntity body = new ByteArrayEntity(t.getBytes(), ContentType.create("text/plain"));
 638             post.setEntity(body);
 639             if (logger.isDebugEnabled()) {
 640                 logger.debug("Requesting records: " + t);
 641             }
 642
 643             HttpResponse response = httpclient.execute(post);
 644             int code = response.getStatusLine().getStatusCode();
 645             if (code != HttpServletResponse.SC_OK) {
 646                 logger.warn("PROV5010: replicateDRLogs failed, bad error code: " + code);
 647                 return;
 648             }
 649             HttpEntity entity = response.getEntity();
 650             String ctype = entity.getContentType().getValue().trim();
 651             if (!ctype.equals("text/plain")) {
 652                 logger.warn("PROV5011: replicateDRLogs failed, bad content type: " + ctype);
 653                 return;
 654             }
 655
 656             String spoolname = "" + System.currentTimeMillis();
 657             Path tmppath = Paths.get(spooldir, spoolname);
 658             Path donepath = Paths.get(spooldir, "IN." + spoolname);
 659             Files.copy(entity.getContent(), Paths.get(spooldir, spoolname), StandardCopyOption.REPLACE_EXISTING);
 660             Files.move(tmppath, donepath, StandardCopyOption.REPLACE_EXISTING);
 661             logger.info("Approximately " + bs.cardinality() + " records replicated.");
 662         } catch (Exception e) {
 663             logger.warn("PROV5012: replicateDRLogs failed, exception: " + e);
 664         } finally {
 665             post.releaseConnection();
 666         }
 667     }
 668 }