alluxio.worker.s3

Class S3NettyHandler

java.lang.Object

alluxio.worker.s3.S3NettyHandler

public class S3NettyHandler
extends Object

S3 Netty Handler - handle http requests from S3 API in Netty.

Field Summary

Fields

Modifier and Type	Field and Description
static Pattern	BUCKET_ADJACENT_DOTS_DASHES_PATTERN
static Pattern	BUCKET_INVALID_SUFFIX_PATTERN
static Pattern	BUCKET_INVALIDATION_PREFIX_PATTERN
static boolean	BUCKET_NAMING_RESTRICTION_ENABLED
static com.google.common.cache.Cache<String,Boolean>	BUCKET_PATH_CACHE
static int	BUCKET_PATH_CACHE_SIZE
static Pattern	BUCKET_PATH_PATTERN
static Pattern	BUCKET_VALID_NAME_PATTERN
AsyncUserAccessAuditLogWriter	mAsyncAuditLogWriter
static Pattern	OBJECT_PATH_PATTERN
static WritePType	S3_WRITE_TYPE

Constructor Summary

Constructors

Constructor and Description
S3NettyHandler(String bucket, String object, io.netty.handler.codec.http.HttpRequest request, io.netty.channel.ChannelHandlerContext ctx, FileSystem fileSystem, DoraWorker doraWorker, AsyncUserAccessAuditLogWriter asyncAuditLogWriter) Constructs an instance of S3NettyHandler.

Method Summary

Modifier and Type	Method and Description
boolean	addContent(io.netty.handler.codec.http.HttpContent content) Adds content to the content queue.
static void	checkPathIsAlluxioDirectory(FileSystem fs, String bucketPath, S3AuditContext auditContext) Check if a path in alluxio is a directory.
S3AuditContext	createAuditContext(String command, String user, String bucket, String object) Creates a S3AuditContext instance.
static S3NettyHandler	createHandler(io.netty.channel.ChannelHandlerContext context, io.netty.handler.codec.http.HttpRequest request, FileSystem fileSystem, DoraWorker doraWorker, AsyncUserAccessAuditLogWriter asyncAuditLogWriter) Create a S3Handler based on the incoming Request.
void	doAuthentication() Do S3 request authentication.
void	extractAMZHeaders() Utility function to help extract x-amz- headers from request.
String	getBucket() Get the bucket name of this request.
io.netty.channel.ChannelHandlerContext	getContext() Get the channel context of this request.
DoraWorker	getDoraWorker() Get dora worker object.
FileSystem	getFileSystemForUser(String user)
FileTransferType	getFileTransferType() Get the FileTransferType of the netty server.
FileSystem	getFsClient() Get system user FileSystem object.
String	getHeader(String headerName) get specified HTTP header value of this request.
String	getHeaderOrDefault(String headerName, String defaultHeaderValue) get specified HTTP header with a default if not exist.
String	getHttpMethod() get HTTP verb of this request.
io.netty.handler.codec.http.HttpContent	getLatestContent() get HTTP content of this request.
String	getObject() Get the object name of this request.
String	getQueryParameter(String queryParam) retrieve given query parameter value.
io.netty.handler.codec.http.HttpRequest	getRequest() get HTTP request.
io.netty.handler.codec.http.HttpResponse	getResponse() get HTTP response.
S3NettyBaseTask	getS3Task() get S3Task of this S3Handler.
com.google.common.base.Stopwatch	getStopwatch() Get Stopwatch object used for recording this request's latency.
AlluxioURI	getUfsPath(AlluxioURI objectPath) Gets UFS full path from Alluxio path.
String	getUser() Get the user name of this request.
void	init() Initialize the S3Handler object in preparation for handling the request.
static void	logAccess(io.netty.handler.codec.http.HttpRequest request, io.netty.handler.codec.http.HttpResponse response, com.google.common.base.Stopwatch stopWatch, S3NettyBaseTask.OpType opType) Log the access of every single http request.
BlockReader	openBlock(String ufsFullPath, long offset, long length) Gets a BlockReader according the ufs full path, offset and length.
void	processHttpResponse(io.netty.handler.codec.http.HttpResponse response) Writes HttpResponse into context channel, After writes context channel will close.
void	processHttpResponse(io.netty.handler.codec.http.HttpResponse response, boolean closeAfterWrite) Writes HttpResponse into context channel.
void	processMappedResponse(BlockReader blockReader, long objectSize) Writes data into netty channel by copying through ByteBuf.
void	processTransferResponse(DataBuffer packet) Writes a DataBuffer into netty channel.
void	rejectUnsupportedResources() Reject unsupported request from the given subresources from request.
boolean	remainContent()
static void	setEntityTag(FileSystem fs, AlluxioURI objectUri, String entityTag) This helper method is used to set the ETag xAttr on an object.
void	setResponse(io.netty.handler.codec.http.HttpResponse response) set FullHttpResponse for this request.
void	setS3Task(S3NettyBaseTask task) set S3Task for this S3Handler.
void	setStopwatch(com.google.common.base.Stopwatch stopwatch) Set the Stopwatch object used for recording this request's latency.

Methods inherited from class java.lang.Object

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

Field Detail

mAsyncAuditLogWriter

public AsyncUserAccessAuditLogWriter mAsyncAuditLogWriter

BUCKET_PATH_PATTERN

public static final Pattern BUCKET_PATH_PATTERN

OBJECT_PATH_PATTERN

public static final Pattern OBJECT_PATH_PATTERN

BUCKET_NAMING_RESTRICTION_ENABLED

public static final boolean BUCKET_NAMING_RESTRICTION_ENABLED

S3_WRITE_TYPE

public static final WritePType S3_WRITE_TYPE

BUCKET_ADJACENT_DOTS_DASHES_PATTERN

public static final Pattern BUCKET_ADJACENT_DOTS_DASHES_PATTERN

BUCKET_INVALIDATION_PREFIX_PATTERN

public static final Pattern BUCKET_INVALIDATION_PREFIX_PATTERN

BUCKET_INVALID_SUFFIX_PATTERN

public static final Pattern BUCKET_INVALID_SUFFIX_PATTERN

BUCKET_VALID_NAME_PATTERN

public static final Pattern BUCKET_VALID_NAME_PATTERN

BUCKET_PATH_CACHE_SIZE

public static final int BUCKET_PATH_CACHE_SIZE

See Also:

Constant Field Values

BUCKET_PATH_CACHE

public static final com.google.common.cache.Cache<String,Boolean> BUCKET_PATH_CACHE

Constructor Detail

S3NettyHandler

public S3NettyHandler(String bucket,
                      String object,
                      io.netty.handler.codec.http.HttpRequest request,
                      io.netty.channel.ChannelHandlerContext ctx,
                      FileSystem fileSystem,
                      DoraWorker doraWorker,
                      AsyncUserAccessAuditLogWriter asyncAuditLogWriter)

Constructs an instance of S3NettyHandler.

Parameters:

bucket -

object -

request -

ctx -

fileSystem -

doraWorker -

asyncAuditLogWriter -

Method Detail

createHandler

public static S3NettyHandler createHandler(io.netty.channel.ChannelHandlerContext context,
                                           io.netty.handler.codec.http.HttpRequest request,
                                           FileSystem fileSystem,
                                           DoraWorker doraWorker,
                                           AsyncUserAccessAuditLogWriter asyncAuditLogWriter)
                                    throws Exception

Create a S3Handler based on the incoming Request.

Parameters:

context -

request -

fileSystem -

doraWorker -

asyncAuditLogWriter -

Returns:

A S3Handler

Throws:

Exception

init

public void init()
          throws Exception

Initialize the S3Handler object in preparation for handling the request.

Throws:

Exception

extractAMZHeaders

public void extractAMZHeaders()

Utility function to help extract x-amz- headers from request.

rejectUnsupportedResources

public void rejectUnsupportedResources()
                                throws S3Exception

Reject unsupported request from the given subresources from request.

Throws:

S3Exception

doAuthentication

public void doAuthentication()
                      throws Exception

Do S3 request authentication.

Throws:

Exception

createAuditContext

public S3AuditContext createAuditContext(String command,
                                         String user,
                                         @Nullable
                                         String bucket,
                                         @Nullable
                                         String object)

Creates a S3AuditContext instance.

Parameters:

command - the command to be logged by this S3AuditContext

user - user name

bucket - bucket name

object - object name

Returns:

newly-created S3AuditContext instance

processHttpResponse

public void processHttpResponse(io.netty.handler.codec.http.HttpResponse response)

Writes HttpResponse into context channel, After writes context channel will close.

Parameters:

response - HttpResponse object

processHttpResponse

public void processHttpResponse(io.netty.handler.codec.http.HttpResponse response,
                                boolean closeAfterWrite)

Writes HttpResponse into context channel.

Parameters:

response - HttpResponse object

closeAfterWrite - if true, After writes context channel will close

processTransferResponse

public void processTransferResponse(DataBuffer packet)

Writes a DataBuffer into netty channel. It supports zero copy through ByteBuf and FileRegion.

Parameters:

packet - DataBuffer packet

processMappedResponse

public void processMappedResponse(BlockReader blockReader,
                                  long objectSize)
                           throws IOException

Writes data into netty channel by copying through ByteBuf.

Parameters:

blockReader - reader instance

objectSize - size of the object

Throws:

IOException

openBlock

public BlockReader openBlock(String ufsFullPath,
                             long offset,
                             long length)
                      throws IOException,
                             AccessControlException

Gets a BlockReader according the ufs full path, offset and length.

Parameters:

ufsFullPath - UFS full path

offset - the offset of this reading

length - the length of this reading

Returns:

a BlockReader

Throws:

IOException

AccessControlException

setS3Task

public void setS3Task(S3NettyBaseTask task)

set S3Task for this S3Handler.

Parameters:

task -

setResponse

public void setResponse(io.netty.handler.codec.http.HttpResponse response)

set FullHttpResponse for this request.

Parameters:

response -

getUser

public String getUser()

Get the user name of this request.

Returns:

user name

getBucket

public String getBucket()

Get the bucket name of this request.

Returns:

bucket name

getObject

public String getObject()

Get the object name of this request.

Returns:

object name

getContext

public io.netty.channel.ChannelHandlerContext getContext()

Get the channel context of this request.

Returns:

ChannelHandlerContext

getFileTransferType

public FileTransferType getFileTransferType()

Get the FileTransferType of the netty server.

Returns:

ChannelHandlerContext

getS3Task

public S3NettyBaseTask getS3Task()

get S3Task of this S3Handler.

Returns:

S3BaseTask

getFsClient

public FileSystem getFsClient()

Get system user FileSystem object.

Returns:

FileSystem object

getDoraWorker

public DoraWorker getDoraWorker()

Get dora worker object.

Returns:

DoraWorker object

getRequest

public io.netty.handler.codec.http.HttpRequest getRequest()

get HTTP request.

Returns:

HTTP request

getResponse

public io.netty.handler.codec.http.HttpResponse getResponse()

get HTTP response.

Returns:

HTTP response

getLatestContent

public io.netty.handler.codec.http.HttpContent getLatestContent()
                                                         throws InterruptedException

get HTTP content of this request.

Returns:

HTTP content

Throws:

InterruptedException

remainContent

public boolean remainContent()

Returns:

Does there exists content in the content queue

addContent

public boolean addContent(io.netty.handler.codec.http.HttpContent content)

Adds content to the content queue.

Parameters:

content -

Returns:

the result of adding content

getHttpMethod

public String getHttpMethod()

get HTTP verb of this request.

Returns:

HTTP Verb

getHeader

public String getHeader(String headerName)

get specified HTTP header value of this request.

Parameters:

headerName -

Returns:

header value

getHeaderOrDefault

public String getHeaderOrDefault(String headerName,
                                 String defaultHeaderValue)

get specified HTTP header with a default if not exist.

Parameters:

headerName -

defaultHeaderValue -

Returns:

header value

getQueryParameter

@Nullable
public String getQueryParameter(String queryParam)

retrieve given query parameter value.

Parameters:

queryParam -

Returns:

query parameter value

getStopwatch

public com.google.common.base.Stopwatch getStopwatch()

Get Stopwatch object used for recording this request's latency.

Returns:

Stopwatch object

setStopwatch

public void setStopwatch(com.google.common.base.Stopwatch stopwatch)

Set the Stopwatch object used for recording this request's latency.

Parameters:

stopwatch -

getFileSystemForUser

public FileSystem getFileSystemForUser(String user)

Parameters:

user - the Subject name of the filesystem user

Returns:

A FileSystem with the subject set to the provided user

getUfsPath

public AlluxioURI getUfsPath(AlluxioURI objectPath)
                      throws S3Exception

Gets UFS full path from Alluxio path.

Parameters:

objectPath - the Alluxio path

Returns:

UfsBaseFileSystem based full path

Throws:

S3Exception

checkPathIsAlluxioDirectory

public static void checkPathIsAlluxioDirectory(FileSystem fs,
                                               String bucketPath,
                                               @Nullable
                                               S3AuditContext auditContext)
                                        throws S3Exception

Check if a path in alluxio is a directory.

Parameters:

fs - instance of FileSystem

bucketPath - bucket complete path

auditContext - the audit context for exception

Throws:

S3Exception

setEntityTag

public static void setEntityTag(FileSystem fs,
                                AlluxioURI objectUri,
                                String entityTag)
                         throws IOException,
                                AlluxioException

This helper method is used to set the ETag xAttr on an object.

Parameters:

fs - The FileSystem used to make the gRPC request

objectUri - The AlluxioURI for the object to update

entityTag - The entity tag of the object (MD5 checksum of the object contents)

Throws:

IOException

AlluxioException

logAccess

public static void logAccess(io.netty.handler.codec.http.HttpRequest request,
                             io.netty.handler.codec.http.HttpResponse response,
                             com.google.common.base.Stopwatch stopWatch,
                             S3NettyBaseTask.OpType opType)

Log the access of every single http request.

Parameters:

request -

response -

stopWatch -

opType -