Project

General

Profile

1 6398 cjones
/**
2
 *  '$RCSfile$'
3
 *    Purpose: Implements a service for managing a Hazelcast cluster member
4
 *  Copyright: 2011 Regents of the University of California and the
5
 *             National Center for Ecological Analysis and Synthesis
6
 *    Authors: Christopher Jones
7
 *
8
 *   '$Author$'
9
 *     '$Date$'
10
 * '$Revision$'
11
 *
12
 * This program is free software; you can redistribute it and/or modify
13
 * it under the terms of the GNU General Public License as published by
14
 * the Free Software Foundation; either version 2 of the License, or
15
 * (at your option) any later version.
16
 *
17
 * This program is distributed in the hope that it will be useful,
18
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
19
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20
 * GNU General Public License for more details.
21
 *
22
 * You should have received a copy of the GNU General Public License
23
 * along with this program; if not, write to the Free Software
24
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
25
 */
26
27 6399 leinfelder
package edu.ucsb.nceas.metacat.dataone.hazelcast;
28 6398 cjones
29 6489 leinfelder
import java.io.FileNotFoundException;
30 6894 cjones
import java.sql.SQLException;
31 6459 leinfelder
import java.util.Collection;
32
import java.util.Date;
33 7098 leinfelder
import java.util.List;
34 6859 cjones
import java.util.concurrent.locks.Lock;
35 6459 leinfelder
36 6398 cjones
import org.apache.log4j.Logger;
37 6904 cjones
import org.dataone.service.exceptions.InvalidSystemMetadata;
38 6446 leinfelder
import org.dataone.service.types.v1.Identifier;
39
import org.dataone.service.types.v1.Node;
40
import org.dataone.service.types.v1.NodeReference;
41
import org.dataone.service.types.v1.SystemMetadata;
42 6398 cjones
43 6401 cjones
import com.hazelcast.config.Config;
44 6437 leinfelder
import com.hazelcast.config.FileSystemXmlConfig;
45 6446 leinfelder
import com.hazelcast.core.EntryEvent;
46
import com.hazelcast.core.EntryListener;
47 6398 cjones
import com.hazelcast.core.Hazelcast;
48 6711 cjones
import com.hazelcast.core.HazelcastInstance;
49 6446 leinfelder
import com.hazelcast.core.IMap;
50 6887 cjones
import com.hazelcast.core.ISet;
51 6446 leinfelder
import com.hazelcast.core.Member;
52 7098 leinfelder
import com.hazelcast.core.MembershipEvent;
53
import com.hazelcast.core.MembershipListener;
54 6446 leinfelder
import com.hazelcast.partition.Partition;
55
import com.hazelcast.partition.PartitionService;
56 6459 leinfelder
import com.hazelcast.query.EntryObject;
57
import com.hazelcast.query.Predicate;
58
import com.hazelcast.query.PredicateBuilder;
59 6398 cjones
60 6446 leinfelder
import edu.ucsb.nceas.metacat.IdentifierManager;
61
import edu.ucsb.nceas.metacat.McdbDocNotFoundException;
62
import edu.ucsb.nceas.metacat.dataone.D1NodeService;
63 6401 cjones
import edu.ucsb.nceas.metacat.properties.PropertyService;
64 6398 cjones
import edu.ucsb.nceas.metacat.shared.BaseService;
65
import edu.ucsb.nceas.metacat.shared.ServiceException;
66 7098 leinfelder
import edu.ucsb.nceas.metacat.util.DocumentUtil;
67 6489 leinfelder
import edu.ucsb.nceas.utilities.FileUtil;
68 6446 leinfelder
import edu.ucsb.nceas.utilities.PropertyNotFoundException;
69 6398 cjones
/**
70
 * The Hazelcast service enables Metacat as a Hazelcast cluster member
71
 */
72
public class HazelcastService extends BaseService
73 7098 leinfelder
  implements EntryListener<Identifier, SystemMetadata>, MembershipListener {
74 6398 cjones
75 6490 leinfelder
  private static final String SINCE_PROPERTY = "dateSysMetadataModified";
76 6459 leinfelder
77
/* The instance of the logging class */
78 6398 cjones
  private static Logger logMetacat = Logger.getLogger(HazelcastService.class);
79
80
  /* The singleton instance of the hazelcast service */
81 6401 cjones
  private static HazelcastService hzService = null;
82 6398 cjones
83 6401 cjones
  /* The Hazelcast configuration */
84
  private Config hzConfig;
85 6398 cjones
86 6446 leinfelder
  /* The instance of the Hazelcast client */
87 6483 cjones
//  private HazelcastClient hzClient;
88 6446 leinfelder
89
  /* The name of the DataONE Hazelcast cluster group */
90
  private String groupName;
91
92
  /* The name of the DataONE Hazelcast cluster password */
93
  private String groupPassword;
94
95
  /* The name of the DataONE Hazelcast cluster IP addresses */
96
  private String addressList;
97
98
  /* The name of the node map */
99
  private String nodeMap;
100
101
  /* The name of the system metadata map */
102
  private String systemMetadataMap;
103
104
  /* The Hazelcast distributed task id generator namespace */
105
  private String taskIds;
106
107 6462 cjones
  /* The Hazelcast distributed node map */
108 6446 leinfelder
  private IMap<NodeReference, Node> nodes;
109
110
  /* The Hazelcast distributed system metadata map */
111
  private IMap<Identifier, SystemMetadata> systemMetadata;
112 6711 cjones
113 6887 cjones
  /* The name of the identifiers set */
114
  private String identifiersSet;
115
116
  /* The Hazelcast distributed identifiers set */
117
  private ISet<Identifier> identifiers;
118
119 6711 cjones
  private HazelcastInstance hzInstance;
120 6702 cjones
121 6398 cjones
  /*
122
   * Constructor: Creates an instance of the hazelcast service. Since
123
   * this uses a singleton pattern, use getInstance() to gain the instance.
124
   */
125
  private HazelcastService() {
126
127
    super();
128
    _serviceName="HazelcastService";
129
130
    try {
131 6401 cjones
      init();
132 6398 cjones
133
    } catch (ServiceException se) {
134 6437 leinfelder
      logMetacat.error("There was a problem creating the HazelcastService. " +
135 6398 cjones
                       "The error message was: " + se.getMessage());
136
137
    }
138
139
  }
140
141
  /**
142
   *  Get the instance of the HazelcastService that has been instantiated,
143
   *  or instantiate one if it has not been already.
144
   *
145
   * @return hazelcastService - The instance of the hazelcast service
146
   */
147
  public static HazelcastService getInstance(){
148
149 6401 cjones
    if ( hzService == null ) {
150 6398 cjones
151 6401 cjones
      hzService = new HazelcastService();
152 6398 cjones
153
    }
154 6401 cjones
    return hzService;
155 6398 cjones
  }
156
157
  /**
158 6401 cjones
   * Initializes the Hazelcast service
159 6398 cjones
   */
160 6401 cjones
  public void init() throws ServiceException {
161 6398 cjones
162 6963 leinfelder
    logMetacat.debug("HazelcastService.init() called.");
163 6398 cjones
164 6489 leinfelder
	String configFileName = null;
165
	Config config = null;
166
	try {
167
		configFileName = PropertyService.getProperty("dataone.hazelcast.configFilePath");
168
		config = new FileSystemXmlConfig(configFileName);
169
	} catch (Exception e) {
170
		configFileName = PropertyService.CONFIG_FILE_DIR + FileUtil.getFS() + "hazelcast.xml";
171 6963 leinfelder
		logMetacat.warn("Custom Hazelcast configuration not defined, using default: " + configFileName);
172 6489 leinfelder
		// make sure we have the config
173
		try {
174
			config = new FileSystemXmlConfig(configFileName);
175
		} catch (FileNotFoundException e1) {
176
			String msg = e.getMessage();
177
			logMetacat.error(msg);
178
			throw new ServiceException(msg);
179
		}
180
	}
181
182
	Hazelcast.init(config);
183 6711 cjones
  this.hzInstance = Hazelcast.getDefaultInstance();
184
185 6446 leinfelder
    // Get configuration properties on instantiation
186
    try {
187
      groupName =
188
        PropertyService.getProperty("dataone.hazelcast.processCluster.groupName");
189
      groupPassword =
190
        PropertyService.getProperty("dataone.hazelcast.processCluster.password");
191
      addressList =
192
        PropertyService.getProperty("dataone.hazelcast.processCluster.instances");
193
      systemMetadataMap =
194
        PropertyService.getProperty("dataone.hazelcast.storageCluster.systemMetadataMap");
195 6887 cjones
      identifiersSet = PropertyService.getProperty("dataone.hazelcast.storageCluster.identifiersSet");
196
//    nodeMap =
197
//    PropertyService.getProperty("dataone.hazelcast.processCluster.nodesMap");
198 6446 leinfelder
      // Become a DataONE-process cluster client
199 6483 cjones
//      String[] addresses = addressList.split(",");
200
//      hzClient =
201
//        HazelcastClient.newHazelcastClient(this.groupName, this.groupPassword, addresses);
202
//      nodes = hzClient.getMap(nodeMap);
203 6446 leinfelder
204
      // Get a reference to the shared system metadata map as a cluster member
205 7054 leinfelder
      // NOTE: this loads the map from the backing store and can take a long time for large collections
206 6446 leinfelder
      systemMetadata = Hazelcast.getMap(systemMetadataMap);
207
208 6887 cjones
      // Get a reference to the shared identifiers set as a cluster member
209
      identifiers = Hazelcast.getSet(identifiersSet);
210
211 6446 leinfelder
      // Listen for changes to the system metadata map
212
      systemMetadata.addEntryListener(this, true);
213
214 7098 leinfelder
      // Listen for members added/removed
215
      hzInstance.getCluster().addMembershipListener(this);
216
217 6446 leinfelder
    } catch (PropertyNotFoundException e) {
218
219
      String msg = "Couldn't find Hazelcast properties for the DataONE clusters. " +
220
        "The error message was: " + e.getMessage();
221
      logMetacat.error(msg);
222
223 6401 cjones
    }
224 6459 leinfelder
225
    // make sure we have all metadata locally
226
    try {
227
	    // add index for resynch() method
228 6490 leinfelder
    	// can only be added once, TODO: figure out how this works
229
	    //systemMetadata.addIndex(SINCE_PROPERTY, true);
230
		//resynch();
231 6459 leinfelder
	} catch (Exception e) {
232
		String msg = "Problem synchronizing system metadata. " + e.getMessage();
233
		logMetacat.error(msg, e);
234
	}
235 6446 leinfelder
236 6398 cjones
  }
237
238 6462 cjones
  /**
239
   * Get the system metadata map
240
   *
241
   * @return systemMetadata - the hazelcast map of system metadata
242 6703 cjones
   * @param identifier - the identifier of the object as a string
243 6462 cjones
   */
244 6446 leinfelder
  public IMap<Identifier,SystemMetadata> getSystemMetadataMap() {
245
	  return systemMetadata;
246
  }
247 6714 leinfelder
248
  /**
249 6887 cjones
   * Get the identifiers set
250
   * @return identifiers - the set of unique DataONE identifiers in the cluster
251
   */
252
  public ISet<Identifier> getIdentifiers() {
253
      return identifiers;
254
255
  }
256
257
  /**
258 6714 leinfelder
   * When Metacat changes the underlying store, we need to refresh the
259
   * in-memory representation of it.
260
   * @param guid
261
   */
262
  public void refreshSystemMetadataEntry(String guid) {
263
	Identifier identifier = new Identifier();
264
	identifier.setValue(guid);
265
	// force hazelcast to update system metadata in memory from the store
266
	HazelcastService.getInstance().getSystemMetadataMap().evict(identifier);
267
268
  }
269 6702 cjones
270 6859 cjones
  public Lock getLock(String identifier) {
271 6702 cjones
272 6859 cjones
    Lock lock = null;
273 6702 cjones
274
    try {
275 6711 cjones
        lock = getInstance().getHazelcastInstance().getLock(identifier);
276 6702 cjones
277
    } catch (RuntimeException e) {
278
        logMetacat.info("Couldn't get a lock for identifier " +
279 6703 cjones
            identifier + " !!");
280 6702 cjones
    }
281
    return lock;
282
283
  }
284 6446 leinfelder
285 6462 cjones
  /**
286
   * Get the DataONE hazelcast node map
287
   * @return nodes - the hazelcast map of nodes
288
   */
289 6483 cjones
//  public IMap<NodeReference, Node> getNodesMap() {
290
//	  return nodes;
291
//  }
292 6446 leinfelder
293 6398 cjones
  /**
294
   * Indicate whether or not this service is refreshable.
295
   *
296
   * @return refreshable - the boolean refreshable status
297
   */
298
  public boolean refreshable() {
299 6401 cjones
    // TODO: Determine the consequences of restarting the Hazelcast instance
300 6407 cjones
    // Set this to true if it's okay to drop from the cluster, lose the maps,
301
    // and start back up again
302
    return false;
303 6398 cjones
304
  }
305
306
  /**
307
   * Stop the HazelcastService. When stopped, the service will no longer
308
   * respond to requests.
309
   */
310
  public void stop() throws ServiceException {
311
312
    Hazelcast.getLifecycleService().shutdown();
313
314
  }
315
316 6711 cjones
  public HazelcastInstance getHazelcastInstance() {
317
      return this.hzInstance;
318
319
  }
320 6407 cjones
321
  /**
322
   * Refresh the Hazelcast service by restarting it
323
   */
324 6471 jones
  @Override
325 6401 cjones
  protected void doRefresh() throws ServiceException {
326
327 6407 cjones
    // TODO: verify that the correct config file is still used
328 6401 cjones
    Hazelcast.getLifecycleService().restart();
329 6407 cjones
330 6401 cjones
  }
331 6446 leinfelder
332
  /**
333
	 * Implement the EntryListener interface for Hazelcast, reponding to entry
334
	 * added events in the hzSystemMetadata map. Evaluate the entry and create
335
	 * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)
336
	 *
337
	 * @param event - The EntryEvent that occurred
338
	 */
339 6471 jones
	@Override
340 6446 leinfelder
	public void entryAdded(EntryEvent<Identifier, SystemMetadata> event) {
341 7089 cjones
342
	  logMetacat.info("SystemMetadata entry added event on identifier " +
343
	      event.getKey().getValue());
344 6446 leinfelder
		// handle as update - that method will create if necessary
345
		entryUpdated(event);
346 7089 cjones
347 6446 leinfelder
	}
348 6401 cjones
349 6446 leinfelder
	/**
350
	 * Implement the EntryListener interface for Hazelcast, reponding to entry
351
	 * evicted events in the hzSystemMetadata map.  Evaluate the entry and create
352
	 * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)
353
	 *
354
	 * @param event - The EntryEvent that occurred
355
	 */
356 6471 jones
	@Override
357 6446 leinfelder
	public void entryEvicted(EntryEvent<Identifier, SystemMetadata> event) {
358 7089 cjones
359
      logMetacat.info("SystemMetadata entry evicted event on identifier " +
360
          event.getKey().getValue());
361
362 6889 cjones
	    // ensure identifiers are listed in the hzIdentifiers set
363
      if ( !identifiers.contains(event.getKey()) ) {
364
          identifiers.add(event.getKey());
365
      }
366 6446 leinfelder
367
	}
368
369
	/**
370
	 * Implement the EntryListener interface for Hazelcast, reponding to entry
371
	 * removed events in the hzSystemMetadata map.  Evaluate the entry and create
372
	 * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)
373
	 *
374
	 * @param event - The EntryEvent that occurred
375
	 */
376 6471 jones
	@Override
377 6446 leinfelder
	public void entryRemoved(EntryEvent<Identifier, SystemMetadata> event) {
378 7089 cjones
379
    logMetacat.info("SystemMetadata entry removed event on identifier " +
380
        event.getKey().getValue());
381
382
	  // we typically don't remove objects in Metacat, but can remove System Metadata
383 6648 leinfelder
		IdentifierManager.getInstance().deleteSystemMetadata(event.getValue().getIdentifier().getValue());
384 6889 cjones
385
    // keep the hzIdentifiers set in sync with the systemmetadata table
386
    if ( identifiers.contains(event.getKey()) ) {
387
        identifiers.remove(event.getKey());
388
389
    }
390
391 6446 leinfelder
	}
392
393
	/**
394
	 * Implement the EntryListener interface for Hazelcast, reponding to entry
395
	 * updated events in the hzSystemMetadata map.  Evaluate the entry and create
396
	 * CNReplicationTasks as appropriate (for DATA, METADATA, RESOURCE)
397
	 *
398
	 * @param event - The EntryEvent that occurred
399
	 */
400 6471 jones
	@Override
401 6446 leinfelder
	public void entryUpdated(EntryEvent<Identifier, SystemMetadata> event) {
402
403 7089 cjones
			logMetacat.debug("Entry added/updated to System Metadata map: " +
404
			    event.getKey().getValue());
405 6446 leinfelder
			PartitionService partitionService = Hazelcast.getPartitionService();
406
			Partition partition = partitionService.getPartition(event.getKey());
407
			Member ownerMember = partition.getOwner();
408
			if (!ownerMember.localMember()) {
409
				// need to pull the entry into the local store
410 6459 leinfelder
				saveLocally(event.getValue());
411 6446 leinfelder
			}
412
413
			// TODO evaluate the type of system metadata change, decide if it
414
			// warrants a replication event, what type (DATA, METADATA, RESOURCE),
415
			// iteratively lock the PID, create and submit the tasks, and expect a
416
			// result back. Deal with exceptions.
417 6596 leinfelder
			SystemMetadata sysmeta = event.getValue();
418
			if (sysmeta != null) {
419
				boolean isMetadata = D1NodeService.isScienceMetadata(event.getValue());
420
				// TODO: do we need to do anything explicit here?
421
			}
422 6889 cjones
423
	     // ensure identifiers are listed in the hzIdentifiers set
424
      if ( !identifiers.contains(event.getKey()) ) {
425
          identifiers.add(event.getKey());
426
      }
427
428 6446 leinfelder
	}
429 6459 leinfelder
430
	/**
431
	 * Save SystemMetadata to local store if needed
432
	 * @param sm
433
	 */
434
	private void saveLocally(SystemMetadata sm) {
435
		logMetacat.debug("Saving entry locally: " + sm.getIdentifier().getValue());
436
		try {
437
			if (!IdentifierManager.getInstance().systemMetadataExists(sm.getIdentifier().getValue())) {
438 6997 leinfelder
				IdentifierManager.getInstance().insertSystemMetadata(sm);
439 6904 cjones
440 6459 leinfelder
			} else {
441
				IdentifierManager.getInstance().updateSystemMetadata(sm);
442 6904 cjones
443 6459 leinfelder
			}
444
		} catch (McdbDocNotFoundException e) {
445 6904 cjones
			logMetacat.error("Could not save System Metadata to local store.", e);
446
447 6894 cjones
		} catch (SQLException e) {
448 6904 cjones
	      logMetacat.error("Could not save System Metadata to local store.", e);
449
450
    } catch (InvalidSystemMetadata e) {
451
        logMetacat.error("Could not save System Metadata to local store.", e);
452
453
    }
454 6459 leinfelder
	}
455
456
	public void resynch() throws Exception {
457
458
		// get the CN that is online
459
		// TODO: do we even need to use a specific CN?
460
		// All the System Metadata records should be available via the shared map
461
//		NodeList allNodes = CNodeService.getInstance().listNodes();
462
//		Node onlineCN = null;
463
//		for (Node node: allNodes.getNodeList()) {
464
//			if (node.getType().equals(NodeType.CN)) {
465
//				if (node.getState().equals(NodeState.UP)) {
466
//					onlineCN = node;
467
//					break;
468
//				}
469
//			}
470
//		}
471
472
		// get the list of items that have changed since X
473
		Date since = IdentifierManager.getInstance().getLastModifiedDate();
474
		EntryObject e = new PredicateBuilder().getEntryObject();
475
		Predicate predicate = e.get(SINCE_PROPERTY).greaterEqual(since);
476
		Collection<SystemMetadata> updatedSystemMetadata = getSystemMetadataMap().values(predicate);
477
		for (SystemMetadata sm: updatedSystemMetadata) {
478
			saveLocally(sm);
479
		}
480
	}
481 6446 leinfelder
482 7098 leinfelder
	/**
483
	 * When there is missing SystemMetadata on the local member,
484
	 * we retrieve it from the shared map and add it to the local
485
	 * backing store for safe keeping.
486
	 */
487
	@Override
488
	public void memberAdded(MembershipEvent event) {
489
		Member member = event.getMember();
490
		logMetacat.debug("Member added to cluster: " + member.getInetSocketAddress());
491
		boolean isLocal = member.localMember();
492
		if (isLocal) {
493
			logMetacat.debug("Member islocal: " + member.getInetSocketAddress());
494
			List<String> localIds = IdentifierManager.getInstance().getLocalIdsWithNoSystemMetadata(true);
495
			if (localIds != null) {
496
				logMetacat.debug("Member missing SystemMetadata entries, count = " + localIds.size());
497
				for (String localId: localIds) {
498
					logMetacat.debug("Processing system metadata for localId: " + localId);
499
					try {
500
						String docid = DocumentUtil.getSmartDocId(localId);
501
						int rev = DocumentUtil.getRevisionFromAccessionNumber(localId);
502
						String guid = IdentifierManager.getInstance().getGUID(docid, rev);
503
						logMetacat.debug("Found mapped guid: " + guid);
504
						Identifier pid = new Identifier();
505
						pid.setValue(guid);
506
						SystemMetadata sm = systemMetadata.get(pid);
507
						logMetacat.debug("Found shared system metadata for guid: " + guid);
508
						saveLocally(sm);
509
						logMetacat.debug("Saved shared system metadata locally for guid: " + guid);
510
					} catch (Exception e) {
511
						logMetacat.error("Could not save shared SystemMetadata entry locally, localId: " + localId, e);
512
					}
513
				}
514
			}
515
		}
516
	}
517
518
	@Override
519
	public void memberRemoved(MembershipEvent event) {
520
		// TODO Auto-generated method stub
521
522
	}
523
524 6398 cjones
}