Package org.apache.storm.blobstore

Class BlobStore

java.lang.Object

org.apache.storm.blobstore.BlobStore

All Implemented Interfaces:

AutoCloseable, Shutdownable

Direct Known Subclasses:

HdfsBlobStore, LocalFsBlobStore

public abstract class BlobStore
extends Object
implements Shutdownable, AutoCloseable

Provides a way to store blobs that can be downloaded. Blobs must be able to be uploaded and listed from Nimbus, and downloaded from the Supervisors. It is a key value based store. Key being a string and value being the blob data.

ACL checking must take place against the provided subject. If the blob store does not support Security it must validate that all ACLs set are always WORLD, everything.

The users can upload their blobs through the blob store command line. The command line also allows us to update and delete blobs.

Modifying the replication factor only works for HdfsBlobStore as for the LocalFsBlobStore the replication is dependent on the number of Nimbodes available.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected class	BlobStore.BlobStoreFileInputStream	Input stream implementation used for writing both the metadata containing the acl information and the blob data.
protected class	BlobStore.BlobStoreFileOutputStream	Output stream implementation used for reading the metadata and data information.
static class	BlobStore.KeyTranslationIterator	Blob store implements its own version of iterator to list the blobs.

Field Summary

Fields

Modifier and Type	Field	Description
protected static final String	BASE_BLOBS_DIR_NAME

Constructor Summary

Constructors

Constructor	Description
BlobStore()

Method Summary

Modifier and Type	Method	Description
void	close()
void	createBlob(String key, byte[] data, SettableBlobMeta meta, Subject who)	Wrapper called to create the blob which contains the byte data.
void	createBlob(String key, InputStream in, SettableBlobMeta meta, Subject who)	Wrapper called to create the blob which contains the byte data.
abstract AtomicOutputStream	createBlob(String key, SettableBlobMeta meta, Subject who)	Creates the blob.
abstract void	deleteBlob(String key, Subject who)	Deletes the blob data and metadata.
<R> Set<R>	filterAndListKeys(KeyFilter<R> filter)	Filters keys based on the KeyFilter passed as the argument.
abstract InputStreamWithMeta	getBlob(String key, Subject who)	Gets the InputStream to read the blob details.
abstract ReadableBlobMeta	getBlobMeta(String key, Subject who)	Gets the current version of metadata for a blob to be viewed by the user or downloaded by the supervisor.
abstract int	getBlobReplication(String key, Subject who)	Gets the replication factor of the blob.
abstract Iterator<String>	listKeys()	Returns an iterator with all the list of keys currently available on the blob store.
abstract void	prepare(Map<String,Object> conf, String baseDir, NimbusInfo nimbusInfo, ILeaderElector leaderElector)	Allows us to initialize the blob store.
byte[]	readBlob(String key, Subject who)	Wrapper around readBlobTo which returns a ByteArray output stream.
void	readBlobTo(String key, OutputStream out, Subject who)	Reads the blob from the blob store and writes it into the output stream.
abstract void	setBlobMeta(String key, SettableBlobMeta meta, Subject who)	Sets the metadata with renewed acls for the blob.
abstract void	setLeaderElector(ILeaderElector leaderElector)	Sets leader elector (only used by LocalFsBlobStore to help sync blobs between Nimbi.
void	startSyncBlobs()	Start the syncing blobs between the local running instance of the BlobStore and others.
Set<String>	storedTopoIds()	Get IDs stored in blob store.
void	updateBlob(String key, byte[] data, Subject who)	Wrapper called to create the blob which contains the byte data.
abstract AtomicOutputStream	updateBlob(String key, Subject who)	Updates the blob data.
abstract int	updateBlobReplication(String key, int replication, Subject who)	Modifies the replication factor of the blob.
void	updateLastBlobUpdateTime()	Updates the last update time of existing blobs in the blobstore to the current time.
void	validateBlobUpdateTime()	Validates that the blob update time of the blobstore is up to date with the current existing blobs.
static final void	validateKey(String key)	Validates key checking for potentially harmful patterns.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.storm.daemon.Shutdownable

shutdown

Field Details

BASE_BLOBS_DIR_NAME

protected static final String BASE_BLOBS_DIR_NAME

See Also:

• Constant Field Values

Constructor Details

BlobStore

public BlobStore()

Method Details

validateKey

public static final void validateKey(String key)
                              throws IllegalArgumentException

Validates key checking for potentially harmful patterns.

Parameters:

key - Key for the blob

Throws:

IllegalArgumentException

prepare

public abstract void prepare(Map<String,Object> conf,
 String baseDir,
 NimbusInfo nimbusInfo,
 ILeaderElector leaderElector)

Allows us to initialize the blob store.

Parameters:

conf - The storm configuration

baseDir - The directory path to store the blobs

nimbusInfo - Contains the nimbus host, port and leadership information

startSyncBlobs

public void startSyncBlobs()
                    throws KeyNotFoundException,
AuthorizationException

Start the syncing blobs between the local running instance of the BlobStore and others. A no-op for the HdfsBlobStore where HDFS itself does the syncing but for the LocalFsBlobStore ZK state updates are run periodically here.

Throws:

KeyNotFoundException

AuthorizationException

createBlob

public abstract AtomicOutputStream createBlob(String key,
 SettableBlobMeta meta,
 Subject who)
                                       throws AuthorizationException,
KeyAlreadyExistsException

Creates the blob.

Parameters:

key - Key for the blob

meta - Metadata which contains the acls information

who - Is the subject creating the blob

Returns:

AtomicOutputStream returns a stream into which the data can be written

Throws:

AuthorizationException

KeyAlreadyExistsException

createBlob

public void createBlob(String key,
 byte[] data,
 SettableBlobMeta meta,
 Subject who)
                throws AuthorizationException,
KeyAlreadyExistsException,
IOException

Wrapper called to create the blob which contains the byte data.

Parameters:

key - Key for the blob

data - Byte data that needs to be uploaded

meta - Metadata which contains the acls information

who - Is the subject creating the blob

Throws:

AuthorizationException

KeyAlreadyExistsException

IOException

createBlob

public void createBlob(String key,
 InputStream in,
 SettableBlobMeta meta,
 Subject who)
                throws AuthorizationException,
KeyAlreadyExistsException,
IOException

Wrapper called to create the blob which contains the byte data.

Parameters:

key - Key for the blob

in - InputStream from which the data is read to be written as a part of the blob

meta - Metadata which contains the acls information

who - Is the subject creating the blob

Throws:

AuthorizationException

KeyAlreadyExistsException

IOException

updateBlob

public abstract AtomicOutputStream updateBlob(String key,
 Subject who)
                                       throws AuthorizationException,
KeyNotFoundException

Updates the blob data.

Parameters:

key - Key for the blob

who - Is the subject having the write privilege for the blob

Returns:

AtomicOutputStream returns a stream into which the data can be written

Throws:

AuthorizationException

KeyNotFoundException

updateBlob

public void updateBlob(String key,
 byte[] data,
 Subject who)
                throws AuthorizationException,
IOException,
KeyNotFoundException

Wrapper called to create the blob which contains the byte data.

Parameters:

key - Key for the blob

data - Byte data that needs to be uploaded

who - Is the subject creating the blob

Throws:

AuthorizationException

IOException

KeyNotFoundException

getBlobMeta

public abstract ReadableBlobMeta getBlobMeta(String key,
 Subject who)
                                      throws AuthorizationException,
KeyNotFoundException

Gets the current version of metadata for a blob to be viewed by the user or downloaded by the supervisor.

Parameters:

key - Key for the blob

who - Is the subject having the read privilege for the blob

Returns:

AtomicOutputStream returns a stream into which the data can be written

Throws:

AuthorizationException

KeyNotFoundException

setLeaderElector

public abstract void setLeaderElector(ILeaderElector leaderElector)

Sets leader elector (only used by LocalFsBlobStore to help sync blobs between Nimbi.

setBlobMeta

public abstract void setBlobMeta(String key,
 SettableBlobMeta meta,
 Subject who)
                          throws AuthorizationException,
KeyNotFoundException

Sets the metadata with renewed acls for the blob.

Parameters:

key - Key for the blob

meta - Metadata which contains the updated acls information

who - Is the subject having the write privilege for the blob

Throws:

AuthorizationException

KeyNotFoundException

deleteBlob

public abstract void deleteBlob(String key,
 Subject who)
                         throws AuthorizationException,
KeyNotFoundException

Deletes the blob data and metadata.

Parameters:

key - Key for the blob

who - Is the subject having write privilege for the blob

Throws:

AuthorizationException

KeyNotFoundException

getBlob

public abstract InputStreamWithMeta getBlob(String key,
 Subject who)
                                     throws AuthorizationException,
KeyNotFoundException

Gets the InputStream to read the blob details.

Parameters:

key - Key for the blob

who - Is the subject having the read privilege for the blob

Returns:

InputStreamWithMeta has the additional file length and version information

Throws:

AuthorizationException

KeyNotFoundException

listKeys

public abstract Iterator<String> listKeys()

Returns an iterator with all the list of keys currently available on the blob store.

Returns:

Iterator<String>

getBlobReplication

public abstract int getBlobReplication(String key,
 Subject who)
                                throws Exception

Gets the replication factor of the blob.

Parameters:

key - Key for the blob

who - Is the subject having the read privilege for the blob

Returns:

BlobReplication object containing the replication factor for the blob

Throws:

Exception

updateBlobReplication

public abstract int updateBlobReplication(String key,
 int replication,
 Subject who)
                                   throws AuthorizationException,
KeyNotFoundException,
IOException

Modifies the replication factor of the blob.

Parameters:

key - Key for the blob

replication - The replication factor the blob has to be set

who - Is the subject having the update privilege for the blob

Returns:

BlobReplication object containing the updated replication factor for the blob

Throws:

AuthorizationException

KeyNotFoundException

IOException

close

public void close()

Specified by:

close in interface AutoCloseable

filterAndListKeys

public <R> Set<R> filterAndListKeys(KeyFilter<R> filter)

Filters keys based on the KeyFilter passed as the argument.

Parameters:

filter - KeyFilter

Returns:

Set of filtered keys

readBlobTo

public void readBlobTo(String key,
 OutputStream out,
 Subject who)
                throws IOException,
KeyNotFoundException,
AuthorizationException

Reads the blob from the blob store and writes it into the output stream.

Parameters:

key - Key for the blob

out - Output stream

who - Is the subject having read privilege for the blob

Throws:

IOException

KeyNotFoundException

AuthorizationException

readBlob

public byte[] readBlob(String key,
 Subject who)
                throws IOException,
KeyNotFoundException,
AuthorizationException

Wrapper around readBlobTo which returns a ByteArray output stream.

Parameters:

key - Key for the blob

who - Is the subject having the read privilege for the blob

Throws:

IOException

KeyNotFoundException

AuthorizationException

storedTopoIds

public Set<String> storedTopoIds()

Get IDs stored in blob store.

Returns:

a set of all of the topology ids with special data stored in the blob store.

updateLastBlobUpdateTime

public void updateLastBlobUpdateTime()
                              throws IOException

Updates the last update time of existing blobs in the blobstore to the current time.

Throws:

IOException - on any error

validateBlobUpdateTime

public void validateBlobUpdateTime()
                            throws IOException

Validates that the blob update time of the blobstore is up to date with the current existing blobs.

Throws:

IOException - on any error