Abstract:
A non-transitory computer readable storage medium has instructions executed by a processor to intercept a query statement at a master machine. The query statement is an instruction from a client machine that specifies how data managed by a distributed storage system should be processed and provided back to the client. In the communication between the client and the master machine, tokens associated with the statement are evaluated to selectively identify a pattern match of one of connection pattern tokens, login pattern tokens or query pattern tokens. For the query pattern tokens, altered tokens for the query statement are formed in response to the pattern match to establish a revised statement. The revised statement is produced in response to application of a policy rule. The revised statement maintains computation, logic and procedure of the statement, but alters parameters of the statement as specified by the policy rule.

Description:
CROSS-REFERENCE TO RELATED APPLICATION 
       [0001]    This application claims priority to U.S. Provisional Patent Application Ser. No. 62/101,341, filed Jan. 8, 2015, the contents of which are incorporated herein by reference. 
     
    
     FIELD OF THE INVENTION 
       [0002]    This invention relates generally to data processing in a network. More particularly, this invention relates to distributed storage and distributed processing statement interception and modification. 
       BACKGROUND OF THE INVENTION 
       [0003]    Query statements can be formed to obtain data from distributed storage and distributed processing resources. The distributed storage may be a distributed database or a distributed file system. Apache Hadoop® is an open-source software framework for distributed storage and distributed processing of very large data sets on computer clusters built from commodity hardware. 
         [0004]    In order to implement distributed computing, there are typically three main functional components: the client component, the management component and the worker component. The client component accepts input from the end user and converts it into requests that are sent to the management component. The management component splits a request into multiple sub-requests and sends the sub-requests to the worker components for processing in parallel. The worker components each process the request by operating on data. Typically, the management component serves as the first point of response serving many requests by many clients simultaneously and coordinating multiple worker components for any given request. In some cases, there may be another machine delegated to receive requests from the client application on behalf of the management component. 
         [0005]    This three-component paradigm is applicable across all major components of Hadoop including the storage layer (e.g., Hadoop Distributed Files System (HDFS)), the compute layer (i.e., MapReduce®) and the access layer (e.g., Hive®, Solr®, Spark®, Storm®, etc.), where there is a server component that responds to the requests from various clients. 
         [0006]    One challenge working with an open source platform, such as Hadoop, is that the code base continuously changes due to contributions from the contributor community. Consequently, it is difficult to augment system functionality in a manner that is guaranteed to work with the continuously changing code base. Accordingly, it would be desirable to develop techniques for producing statement modifications that are certain to be operative with a continuously changing code base. 
       SUMMARY OF THE INVENTION 
       [0007]    A non-transitory computer readable storage medium has instructions executed by a processor to intercept a query statement at a master machine, or a machine delegated by the master machine to receive requests from clients. The query statement is an instruction from a client machine that specifies how data managed by a distributed storage system should be processed and provided back to the client. In the communication between the client and the master machine, tokens associated with the statement are evaluated to selectively identify a pattern match of one of connection pattern token, login pattern tokens or query pattern tokens. For the query pattern tokens, altered tokens for the query statement are formed in response to the pattern match to establish a revised statement. The revised statement is produced in response to application of a policy rule. The revised statement maintains computation, logic and procedure of the statement, but alters parameters of the statement as specified by the policy rule. 
     
    
     
       BRIEF DESCRIPTION OF THE FIGURES 
         [0008]    The invention is more fully appreciated in connection with the following detailed description taken in conjunction with the accompanying drawings, in which: 
           [0009]      FIG. 1  illustrates a system configured in accordance with an embodiment of the invention. 
           [0010]      FIG. 2  illustrates processing operations associated with a proxy and policy engine utilized in accordance with an embodiment of the invention. 
           [0011]      FIG. 3  illustrates processing operations associated with different components of the system. 
           [0012]      FIGS. 4A-4B  illustrate packet processing performed in accordance with an embodiment of the invention. 
           [0013]      FIGS. 5-8  illustrate byte processing performed in accordance with an embodiment of the invention. 
       
    
    
       [0014]    Like reference numerals refer to corresponding parts throughout the several views of the drawings. 
       DETAILED DESCRIPTION OF THE INVENTION 
       [0015]      FIG. 1  illustrates a network  100  configured in accordance with an embodiment of the invention. The network  100  includes a server  102  connected to a set of servers  104 _ 1  through  104 _N via a network  106 . The server  102  can be a master server, while servers  104 _ 1  through  104 _N are worker servers in a distributed storage and a distributed processing environment. Network  106  may be any combination of wired and wireless networks. 
         [0016]    Server  102  includes standard components, such as a central processing unit  110  connected to input/output devices  112  via a bus  114 . The input/output devices  112  may include a keyboard, mouse, touch display and the like. A network interface circuit  116  is also connected to the bus  114  to provide connectivity to network  106 . A memory  120  is also connected to the bus  114 . The memory stores instructions executed by the central processing unit  110  to implement operations of the invention. In particular, the memory  120  stores a master module  122  to implement standard master node operations within a distributed storage and distributed processing environment. The memory also stores a proxy  124  and a policy engine  126  which include instructions executed by the processor  110  to implement operations of the invention. It is also possible for the proxy  124  and policy engine  126  to reside on a separate machine that can communicate with other machines on the network  106 . The proxy  124  effectively sits in front of the master module  122  and continuously listens for request from clients on an IP/Port and forwards requests from the clients to the appropriate IP/Port to the appropriate management component. The proxy  124  forwards all communications in the native request-response protocol without modification, except in certain cases, such as connection requests, authentication requests and requests for data. The proxy  124  is able to distinguish between which packets to forward without modification and which to hold until they can be modified. In such a case, the proxy  124  traps the request and sends the request and the detected meta-data to the policy engine  126  to get a modified policy compliant request back from the policy engine  126 . The kinds of information that the proxy  124  detects may include the query string, the login name, client parameters such as the IP address of the client, etc. The policy engine  126  may use any of these detected parameters to provide instructions to the proxy  124  on how to modify the bytes of the packets. The proxy  124  then inserts the modified request into part of the session that is interrupted within the native request response protocol format and continues the session forward. The management component processes the modified request received from the proxy  124  as if it were obtained from the client. 
         [0017]    The proxy  124  intercepts the application level details for the Hadoop request-response protocol so that additional logic can be inserted into the communication between client and server components without modifying the code on either side. Most of the protocol is unchanged, only specific data packets are captured and only specific method calls are changed. The proxy  124  reconstructs packets to inject desired functionality. 
         [0018]    Each worker server  104 _ 1  through  104 _N also includes standard components, such as a central processing unit  130 , bus  134 , input/output devices  132  and a network interface circuit  136 . A memory  140  is connected to bus  132 . The memory  140  stores a worker module  142  to implement distributed storage and distributed processing operations. Each worker server may incorporate a proxy  124 , a policy engine  126 , or both if additional per worker modification of the query is desired. 
         [0019]    The network  100  may also include a client machine  148 . The client machine  148  includes standard components, such as a central processing unit  150 , input/output devices  152 , a bus  154  and a network interface circuit  156 . A memory  160  is connected to bus  154 . The memory  160  stores a client module  162  with instructions executed by the central processing unit  150 . The client module  162  may facilitate the formation of a query, which is then directed toward the master module  122  of master server  102 . The client module  162  includes drivers to send communications to the master module  122 , where it is intercepted by proxy  124  before the communication is received by the master module  122 , as discussed below. In many cases, the client module  162  and master module  122  are not aware that the proxy  124  is intercepting the communication because neither the client module  162  nor the master module  122  need to change their standard protocol for communication. 
         [0020]      FIG. 2  illustrates processing operations associated with the proxy  124  and policy engine  126 . A statement is intercepted  200 . As previously indicated, the proxy  124  effectively sits in front of the master module  122  to intercepts statements. Tokens (e.g., bytes or combinations of bytes) are evaluated for pattern matches  202 . Tokens are matched for connection patterns, login patterns and query patterns. The policy engine then forms altered tokens to form a revised statement  204 . The altered tokens are altered in order to enforce a specified policy, such as a connection policy, a login policy and/or a data access policy. The packets containing the revised statement are received by the master module  122 . 
         [0021]    The revised statement is then executed  206 . For example, the master module  122  develops a query plan. The query plan is then deployed  208  to worker nodes. Results are combined  210 . That is, the master module  122  combines results received from the worker nodes. The combined results form a response. The response is then supplied  212  to the client. 
         [0022]      FIG. 3  more fully characterizes this processing. A client machine  102  generates a request  300 . The proxy  124  traps the request  302 . That is, the proxy  124  traps the requests if there is a token pattern match. If so, the request is sent to the policy engine  126 , which applies one or more policies  304  to the request to generate an altered request  306 . The altered request is passed to the proxy  124 , which sends the altered request  308  to the master module  122 . The master module processes the altered request  310 . In the case of a data request, the master module devises a query plan  312 . The query plan is then deployed  314  to workers  104 _ 1  through  104 _N. The workers execute the query plan  316  and return results  318 . The master module  122  combines the results to form a response  320 . The response is then supplied  322  to the client machine  102 . The client machine  102  receives the response  324 . 
         [0023]      FIGS. 4A and 4B  illustrate packet processing operations performed in accordance with an embodiment of the invention. A client connection is accepted  400  by the proxy  124 . A client thread is opened  402  on the proxy  124 . A packet is received  404 . The packet is analyzed by the proxy  124  to determine a request type  406 . Block  408  tests whether the request is a database client. If the request is not a database client ( 408 —false), other clients are processed  410  until a termination state  412 , upon which processing is stopped  414 . 
         [0024]    Returning to block  408 , if a database client is implicated ( 408 —true) a connection is made to the master module  122  (or a machine delegated by the master machine to receive connections from the client on behalf of the master machine) on a database IP/port  416 . If a connection does not exist ( 418 —false), processing proceeds to block  412 . If a connection exists ( 418 —true), processing proceeds to block  420  of  FIG. 4B  where tokens are decoded. The decoded tokens are searched for connection patterns  422 . If a match exists ( 424 —true), data byte connection information is altered  426  to enforce a policy. The modified bytes are forwarded to a server component  438  (e.g., master module  122 ) for processing. The processed bytes (e.g., a response) is received from the server  440  and is forwarded to client  442  (e.g., the proxy  124  forwards the response to client module  162 ). This processing is repeated until a session break or channel error  444 . In the event of a session break or channel error  444 , processing is stopped, as shown at block  414  of  FIG. 4A . If there is no session break or channel error, processing returns to block  420 . In some cases, another client of the proxy  124  can send a special command to break the session as in block  444 . 
         [0025]    If a match is not found in block  424  ( 424 —false), the bytes are searched for login patterns  428 . If a match is found ( 430 —true) bytes are altered to enforce a login policy  432  and the result is forwarded to a server component  438 . If a login pattern is not matched ( 430 —false), the bytes are searched for a query pattern  434 . If a query pattern if found the query bytes are modified to enforce a policy  436 . The modified bytes are then forwarded to a server component  438 . 
         [0026]      FIG. 5  illustrates bytes  500  and decoded components including a login pattern  502 . In this example, the login credential (bob@agetak.agetak#123) is inadequate. Consequently, the bytes  600  are altered to preclude login. In this example, the plain text of the altered bytes is a character sequence  602  (aaaaaaaaa.aaaaaaa) that will preclude login at the server. 
         [0027]      FIG. 7  illustrates bytes  700  and equivalent plain text  702 . The plain text includes a data access request (ExecuteStatement . . . from purchasenew). Query bytes  800  are altered to enforce a policy. In this case, access to the “purchasenew” source data is blocked by obscuring the source using xxxxxxxx characters  802 . In addition to the modified string as shown in  FIG. 8 , the query string can also contain sub-strings formatted as comments that are ignored by the master module, but that contain additional text that can be used to watermark the session. 
         [0028]    Below is a table illustrating various Hadoop services, the server that handles the Hadoop service, the types of requests that are handled and the meta-data that is detected in accordance with embodiments of the invention. 
         [0000]    
       
         
               
               
               
               
             
           
               
                   
               
               
                 Hadoop ® 
                   
                   
                   
               
               
                 Service 
                 Server 
                 Request 
                 Detected Meta-data 
               
               
                   
               
             
             
               
                 Storage 
                 Name 
                 HDFS commands on a URI 
                 Fields contained 
               
               
                 HDFS 
                 Node 
                 that return the file contents or 
                 within the file 
               
               
                   
                   
                 operate on one or more files 
                 URI 
               
               
                   
                   
                 specified by the URI 
                 Requesting user 
               
               
                   
                   
                   
                 identity 
               
               
                 Computation 
                 Resource 
                 MapReduce job with a Java ® 
                 Fields contained 
               
               
                 MapReduce ® 
                 Manager 
                 program that operates on one or 
                 within the file 
               
               
                 YARN ® 
                   
                 more files specified by a URI 
                 URI 
               
               
                   
                   
                   
                 Fields serialized in 
               
               
                   
                   
                   
                 the Java program 
               
               
                   
                   
                   
                 Methods names and 
               
               
                   
                   
                   
                 arguments in the Java 
               
               
                   
                   
                   
                 program 
               
               
                   
                   
                   
                 Requesting user 
               
               
                   
                   
                   
                 identity 
               
               
                 Access 
                 HiveServer2, 
                 A query (an hql query or a 
                 Fields contained in 
               
               
                 Hive ® 
                 Impala 
                 search query for example) that 
                 the request 
               
               
                 Impala ® 
                 Daemon, 
                 operates on a pre-defined table 
                 Requesting user 
               
               
                 Solr ® 
                 Solr 
                 or index. 
                 identity 
               
               
                   
               
             
          
         
       
     
         [0029]    In case of HDFS, the management component is the Name Node, the worker component is the Data Node, and the client components can be different software including hdfs dfs jar, webhdfs, httpfs, or a custom client created by a software vendor that initiates a request using the HDFS compatible protocol. 
         [0030]    To protect data in the files from being used directly through HDFS clients, the proxy  124  proxies the Name Node. The Name Node is sent meta-data operations requests by any and all clients that issue HDFS commands. When the proxy  124  detects a URI and an HDFS command, it sends the request meta-data including the user identity making the request from the client to the policy engine  126  to obtain a modified URI. The modified URI is based on the entitlement of the requesting user to the file or fields and records contained within the file of the URI. The modified URI may be empty, may contain a series of URIs with fewer files or alternate files with truncated data. The request with the original command and the modified URI are sent to the Name Node for processing. This approach is used for all requests for files that originate from direct clients to HDFS. 
         [0031]    In case of MapReduce, the management component is the Resource Manager. The worker component comprises the Node Manager in MR 1  or a Node Manager and Application Master in the case of MR 2 , which is commonly referred to as YARN. The client can be the MapReduce client provided as part of the Hadoop common library or a language specific implementation, such as Ruby CLI® created by Amazon Web Services (AWS). The Resource Manager in a Hadoop cluster is responsible for tracking the resources in a cluster and scheduling MapReduce jobs. The Resource Manager is sent MapReduce job requests by any and all clients that submit jobs. 
         [0032]    To enforce access control on MapReduce jobs so the job can only operate on the data that the requester of the job is entitled to use, proxy  124  proxies the Resource Manager. For each job submitted to the Resource Manager before it is processed, the proxy traps the request and sends the job meta-data including the user identity making the request from the client to the policy engine  126  to obtain a modified request. Based on the entitlement of the requesting user to the fields present in the URI specified in the job and to the fields deserialized in the Java program provided with the MapReduce job, the modified URI may be blank, may contain a series of URIs with fewer files or alternate files with truncated data. In this case, only the jobs for users with fields allowed in the jobs that operate on the files containing fields allowed to the user are allowed to continue; the rest are terminated. This approach is used to secure all MapReduce jobs. 
         [0033]    In case of the access layer, an important access method includes interactive query methods implemented as Apache Hive, Cloudera Impala®, Hortonworks Stinger®, and other HQL-on-Hadoop mechanisms. In each of these cases, HiveServer2 can be used as a management component that receives requests in Hive Query Language (HQL), converts them into MapReduce jobs for execution and returns the result to the client components. In the case of an MPP method that can use HQL as the query language (for example Impala), the Query Execution Engine is proxied directly (imapald in case of Impala) rather than just going through the HiveServer. 
         [0034]    To enforce access control on HiveQL requests so that the requesting users get only the data they are entitled to in a format that is compatible with the original request that the client expects, proxy  124  may proxy the HiveServer2. For each HQL request sent to the HiveServer by the client, the proxy  124  traps the request and sends the original query statement and requesting user id to the policy engine  126 . Based on the entitlement of the requesting user to the fields present in the query, the modified query may contain the original fields, dummy values, transformed values, partially transformed values, predicates that filter out records that meet certain criteria, or some combination of these. By doing so the HQL statement contains the same computation, logic and procedure, but only the data elements in a set that the requesting user has access to. 
         [0035]    In one embodiment of the invention, the client module  162  may include drivers that intercept a request and redirect it to the proxy  124 . The drivers override the base driver Java classes of the target database and override the function calls to execute the specified functions before creating a connection to the target database and submitting the call. 
         [0036]    A driver class may be derived from the java.HQL.Driver and hence implement all the functioning of a JDBC Driver. The class may be used to connect to the policy engine  126 . It also reads the values of IP and Ports of various servers through the configuration file. It connects to the remote database by means of the IP address and port provided by the user at the time of passing a URL. 
         [0037]    In one embodiment, the proxy  124  is used to get the statement and PreparedStatement objects. It implements the java.HQL.Connection Interface and redefines some of the methods to get these variables by using the SConnection object. It also defines some of the methods to set the IP and Ports for the Policy Engines. 
         [0038]    The proxy  124  may include a Statement proxy (S Statement) that implements the java.HQL.Statement interface and defines all of its methods to execute an HQL query. It also defines some methods to set the policy engine  126  IP and port and some parameters related to databases. 
         [0039]    The proxy  124  may include a PreparedStatement proxy (SPreparedStatement) that implements the java.HQL.PreparedStatement interface and defines all of its methods to execute an HQL query. It also defines some methods to set the policy engine  126  IP and port and some parameters related to databases. 
         [0040]    The end user using the client module JDBC Drivers is first authenticated at the policy engine  126 . The User passes the username along with the URL to a driver. Then a query is passed to the policy engine  126  that executes the “login” command there, and returns the information that the user is valid or not. After proper authentication the user can execute various queries. 
         [0041]    The query that a user enters is passed to the SStatement and SPreparedStatement for execution. The methods executeQuery and executeUpdate are used for this purpose. These methods internally modify the HQL query to be executed by using the getModifiedHQL method, which uses the getModifiedHQL method of a NativeCommand class. This method passes the query to the policy engine  126 , which returns the modified query and then the query is passed to the database for execution. 
         [0042]    The following table provides exemplary services, service functions, software modules and the functions performed by the software modules. 
         [0000]    
       
         
               
               
               
               
             
           
               
                   
               
               
                 Service 
                 Service Function 
                 Module 
                 Module Function 
               
               
                   
               
             
             
               
                 dshive 
                 Enforcement point for 
                 Intercept 
                 Proxy Hive server at TCP level 
               
               
                   
                 applications that can 
                   
                 to intercept the userid, HQL 
               
               
                   
                 use any Hive client and 
                   
                 requests and session attributes to 
               
               
                   
                 HQL to query data in 
                   
                 send to policy engine and 
               
               
                   
                 Hadoop. 
                   
                 repackage the policy compliant 
               
               
                   
                   
                   
                 query in the session for 
               
               
                   
                   
                   
                 execution against the back-end 
               
               
                   
                   
                   
                 database. Used to 
               
               
                   
                   
                   
                 Intercept connection to detect 
               
               
                   
                   
                   
                 end-user id and HQL request 
               
               
                   
                   
                   
                 Put modified HQL in session 
               
               
                   
                   
                   
                 to return a valid response to 
               
               
                   
                   
                   
                 client 
               
               
                   
                   
                   
                 Listen for connection at a valid 
               
               
                   
                   
                   
                 TCP port and forward the 
               
               
                   
                   
                   
                 packet to correct source 
               
               
                   
                   
                   
                 Output audit logs keyed with a 
               
               
                   
                   
                   
                 session id and with timing for 
               
               
                   
                   
                   
                 login, policy decision and back- 
               
               
                   
                   
                   
                 end query response 
               
               
                 policyengine 
                 Policy engine that 
                 End-user 
                 Bridge the interaction between 
               
               
                   
                 allows granular policies 
                 authentication 
                 the intercept and a source of 
               
               
                   
                 to be enforced by 
                   
                 user-id/password authentication. 
               
               
                   
                 accepting a user identity 
                   
                 Possible sources are external 
               
               
                   
                 and a request in a query 
                   
                 LDAP, external Active 
               
               
                   
                 language (HQL, HQL, 
                   
                 Directory, external databases 
               
               
                   
                 SparQL, etc) and 
                   
                 and internal user store. Used for 
               
               
                   
                 returning a policy 
                   
                 the following capabilities: 
               
               
                   
                 compliant request that 
                   
                 User identity to group 
               
               
                   
                 can be executed against 
                   
                 membership in LDAP or Active 
               
               
                   
                 the physical data source 
                   
                 Directory. 
               
               
                   
                 to get a policy 
                   
                 LDAP over SSL between 
               
               
                   
                 compliant result. 
                   
                 module and external LDAP. 
               
               
                   
                   
                   
                 Policy based connection 
               
               
                   
                   
                   
                 authorization. 
               
               
                   
                   
                 Policy query re- 
                 Reengineer a query on a 
               
               
                   
                   
                 engineering 
                 physical model to make it policy 
               
               
                   
                   
                   
                 compliant. It tokenizes a query, 
               
               
                   
                   
                   
                 runs tokens against ACL to 
               
               
                   
                   
                   
                 make policy decisions and 
               
               
                   
                   
                   
                 constructs a valid HQL query 
               
               
                   
                   
                   
                 that is policy compliant. 
               
               
                   
                   
                 Attribute 
                 For a given user thread, looks 
               
               
                   
                   
                 resolution 
                 up the value of attributes that 
               
               
                   
                   
                   
                 need to be computed against an 
               
               
                   
                   
                   
                 external LDAP or HQL source. 
               
               
                   
               
             
          
         
       
     
         [0043]    An embodiment of the present invention relates to a computer storage product with a non-transitory computer readable storage medium having computer code thereon for performing various computer-implemented operations. The media and computer code may be those specially designed and constructed for the purposes of the present invention, or they may be of the kind well known and available to those having skill in the computer software arts. Examples of computer-readable media include, but are not limited to: magnetic media, optical media, magneto-optical media and hardware devices that are specially configured to store and execute program code, such as application-specific integrated circuits (“ASICs”), programmable logic devices (“PLDs”) and ROM and RAM devices. Examples of computer code include machine code, such as produced by a compiler, and files containing higher-level code that are executed by a computer using an interpreter. For example, an embodiment of the invention may be implemented using JAVA®, C++, or other object-oriented programming language and development tools. Another embodiment of the invention may be implemented in hardwired circuitry in place of, or in combination with, machine-executable software instructions. 
         [0044]    The foregoing description, for purposes of explanation, used specific nomenclature to provide a thorough understanding of the invention. However, it will be apparent to one skilled in the art that specific details are not required in order to practice the invention. Thus, the foregoing descriptions of specific embodiments of the invention are presented for purposes of illustration and description. They are not intended to be exhaustive or to limit the invention to the precise forms disclosed; obviously, many modifications and variations are possible in view of the above teachings. The embodiments were chosen and described in order to best explain the principles of the invention and its practical applications, they thereby enable others skilled in the art to best utilize the invention and various embodiments with various modifications as are suited to the particular use contemplated. It is intended that the following claims and their equivalents define the scope of the invention.