You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by al...@apache.org on 2012/11/19 19:39:31 UTC
git commit: add basic authentication support for Pig CassandraStorage;
patch by Aleksey Yeschenko, reviewed by Brandon Williams for CASSANDRA-3042
Updated Branches:
refs/heads/cassandra-1.1 e30519f33 -> fab61e309
add basic authentication support for Pig CassandraStorage; patch by
Aleksey Yeschenko, reviewed by Brandon Williams for CASSANDRA-3042
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/fab61e30
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/fab61e30
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/fab61e30
Branch: refs/heads/cassandra-1.1
Commit: fab61e309fd4ddf312f033f3458a9aa067279090
Parents: e30519f
Author: Aleksey Yeschenko <al...@apache.org>
Authored: Mon Nov 19 21:37:09 2012 +0300
Committer: Aleksey Yeschenko <al...@apache.org>
Committed: Mon Nov 19 21:37:57 2012 +0300
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../org/apache/cassandra/hadoop/ConfigHelper.java | 22 ++++
.../cassandra/hadoop/pig/CassandraStorage.java | 86 ++++++++++-----
3 files changed, 83 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fab61e30/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ef5e43b..a89184e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.1.7
+ * add basic authentication support for Pig CassandraStorage (CASSANDRA-3042)
* fix CQL2 ALTER TABLE compaction_strategy_class altering (CASSANDRA-4965)
* reset getRangeSlice filter after finishing a row for get_paged_slice
(CASSANDRA-4919)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fab61e30/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
index 1646635..4b49387 100644
--- a/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
+++ b/src/java/org/apache/cassandra/hadoop/ConfigHelper.java
@@ -295,16 +295,38 @@ public class ConfigHelper
return conf.get(OUTPUT_KEYSPACE_CONFIG);
}
+ public static void setInputKeyspaceUserNameAndPassword(Configuration conf, String username, String password)
+ {
+ setInputKeyspaceUserName(conf, username);
+ setInputKeyspacePassword(conf, password);
+ }
+
+ public static void setInputKeyspaceUserName(Configuration conf, String username)
+ {
+ conf.set(INPUT_KEYSPACE_USERNAME_CONFIG, username);
+ }
+
public static String getInputKeyspaceUserName(Configuration conf)
{
return conf.get(INPUT_KEYSPACE_USERNAME_CONFIG);
}
+ public static void setInputKeyspacePassword(Configuration conf, String password)
+ {
+ conf.set(INPUT_KEYSPACE_PASSWD_CONFIG, password);
+ }
+
public static String getInputKeyspacePassword(Configuration conf)
{
return conf.get(INPUT_KEYSPACE_PASSWD_CONFIG);
}
+ public static void setOutputKeyspaceUserNameAndPassword(Configuration conf, String username, String password)
+ {
+ setOutputKeyspaceUserName(conf, username);
+ setOutputKeyspacePassword(conf, password);
+ }
+
public static void setOutputKeyspaceUserName(Configuration conf, String username)
{
conf.set(OUTPUT_KEYSPACE_USERNAME_CONFIG, username);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/fab61e30/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
index 8f539a9..c2f1c13 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
@@ -21,38 +21,31 @@ import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.*;
-import org.apache.cassandra.config.ConfigurationException;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.marshal.IntegerType;
-import org.apache.cassandra.db.marshal.TypeParser;
-import org.apache.cassandra.thrift.*;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Hex;
-import org.apache.cassandra.utils.UUIDGen;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.Column;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.marshal.AbstractCompositeType.CompositeComponent;
import org.apache.cassandra.hadoop.*;
-import org.apache.cassandra.thrift.Mutation;
-import org.apache.cassandra.thrift.Deletion;
-import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Hex;
+import org.apache.cassandra.utils.UUIDGen;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.*;
-
import org.apache.pig.*;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.*;
-import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.impl.util.UDFContext;
+import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
@@ -93,6 +86,8 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
private ByteBuffer slice_end = BOUND;
private boolean slice_reverse = false;
private boolean allow_deletes = false;
+ private String username;
+ private String password;
private String keyspace;
private String column_family;
private String loadSignature;
@@ -467,8 +462,6 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
private void setLocationFromUri(String location) throws IOException
{
- // parse uri into keyspace and columnfamily
- String names[];
try
{
if (!location.startsWith("cassandra://"))
@@ -496,12 +489,23 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
usePartitionFilter = Boolean.parseBoolean(urlQuery.get("use_secondary"));
}
String[] parts = urlParts[0].split("/+");
- keyspace = parts[1];
+ String[] credentialsAndKeyspace = parts[1].split("@");
+ if (credentialsAndKeyspace.length > 1)
+ {
+ String[] credentials = credentialsAndKeyspace[0].split(":");
+ username = credentials[0];
+ password = credentials[1];
+ keyspace = credentialsAndKeyspace[1];
+ }
+ else
+ {
+ keyspace = parts[1];
+ }
column_family = parts[2];
}
catch (Exception e)
{
- throw new IOException("Expected 'cassandra://<keyspace>/<columnfamily>[?slice_start=<start>&slice_end=<end>[&reversed=true][&limit=1][&allow_deletes=true][widerows=true][use_secondary=true]]': " + e.getMessage());
+ throw new IOException("Expected 'cassandra://[username:password@]<keyspace>/<columnfamily>[?slice_start=<start>&slice_end=<end>[&reversed=true][&limit=1][&allow_deletes=true][widerows=true][use_secondary=true]]': " + e.getMessage());
}
}
@@ -559,6 +563,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
{
conf = job.getConfiguration();
setLocationFromUri(location);
+
if (ConfigHelper.getInputSlicePredicate(conf) == null)
{
SliceRange range = new SliceRange(slice_start, slice_end, slice_reverse, limit);
@@ -573,6 +578,9 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
if (usePartitionFilter && getIndexExpressions() != null)
ConfigHelper.setInputRange(conf, getIndexExpressions());
+ if (username != null && password != null)
+ ConfigHelper.setInputKeyspaceUserNameAndPassword(conf, username, password);
+
ConfigHelper.setInputColumnFamily(conf, keyspace, column_family, widerows);
setConnectionInformation();
@@ -800,6 +808,10 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
{
conf = job.getConfiguration();
setLocationFromUri(location);
+
+ if (username != null && password != null)
+ ConfigHelper.setOutputKeyspaceUserNameAndPassword(conf, username, password);
+
ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
setConnectionInformation();
@@ -1032,18 +1044,38 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
private void initSchema(String signature)
{
- UDFContext context = UDFContext.getUDFContext();
- Properties property = context.getUDFProperties(CassandraStorage.class);
+ Properties properties = UDFContext.getUDFContext().getUDFProperties(CassandraStorage.class);
// Only get the schema if we haven't already gotten it
- if (!property.containsKey(signature))
+ if (!properties.containsKey(signature))
{
- Cassandra.Client client = null;
try
{
- client = ConfigHelper.getClientFromInputAddressList(conf);
- CfDef cfDef = null;
+ Cassandra.Client client = ConfigHelper.getClientFromInputAddressList(conf);
client.set_keyspace(keyspace);
+
+ if (username != null && password != null)
+ {
+ Map<String, String> credentials = new HashMap<String, String>(2);
+ credentials.put(IAuthenticator.USERNAME_KEY, username);
+ credentials.put(IAuthenticator.PASSWORD_KEY, password);
+
+ try
+ {
+ client.login(new AuthenticationRequest(credentials));
+ }
+ catch (AuthenticationException e)
+ {
+ logger.error("Authentication exception: invalid username and/or password");
+ throw new RuntimeException(e);
+ }
+ catch (AuthorizationException e)
+ {
+ throw new AssertionError(e); // never actually throws AuthorizationException.
+ }
+ }
+
+ CfDef cfDef = null;
KsDef ksDef = client.describe_keyspace(keyspace);
List<CfDef> defs = ksDef.getCf_defs();
for (CfDef def : defs)
@@ -1055,9 +1087,11 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
}
}
if (cfDef != null)
- property.setProperty(signature, cfdefToString(cfDef));
+ properties.setProperty(signature, cfdefToString(cfDef));
else
- throw new RuntimeException("Column family '" + column_family + "' not found in keyspace '" + keyspace + "'");
+ throw new RuntimeException(String.format("Column family '%s' not found in keyspace '%s'",
+ column_family,
+ keyspace));
}
catch (TException e)
{