fix lua cross slot, clean up REDIS_URI parameter addition

Author: jphaugla

README.md

@@ -69,10 +69,12 @@ mvn clean verify
 ```
  *  run the consumer   
 ```bash
+export REDIS_URI=redis://localhost:12000
 ./runconsumer.sh
 ```
  * run the producer
 ```bash
+# will produce to localhost 12000 and 12002.   (a little bit odd maybe)
 ./runproducer.sh
 ```
 ### Verify the results
@@ -102,17 +104,21 @@ hgetall weather_sensor:wind:message:MSG2:float
 
 ## run with LUA consumer
 (Alternatively, this can be run through intellij)
-* Compile the code
+* Compile the code.  These will use the REDIS_URI environment variable from setEnvironment.sh
 ```bash
 mvn clean verify
 ```
-*  run the consumer
+* run the consumer
+* These scripts use REDIS_URI environment variable for connectivity
+*NOTE:* this use of one hash key for all the keys is needed because LUA only can function on redis objects in the exact same hash slot. Using this type of technique can severely skew the data so is for demo purpose only.
 ```bash
+export REDIS_URI=redis://localhost:12000
 ./runconsumerLUA.sh
 ```
-*   run the producer
+*   run the producer.  
 ```bash
-./runproducer.sh
+export REDIS_URI=redis://localhost:12000
+./runproducerSingle.sh
 ```
 ### Verify the results
 ```bash
@@ -126,10 +132,12 @@ The individual redis sets and hashes are the same as above in the simple consume
     * The crdpurge.sh is very handy to efficiently do a "purgeall" of the database contents
  * Can run a consumer against each "region" and both will receive all the messages.  The crdb logic will handle de-duplication of the redis objects.
 ```bash
+export REDIS_URI=redis://localhost:12000
  ./runconsumer.sh
  ```
 ```bash
-./runconsumer2.sh
+export REDIS_URI=redis://localhost:12002
+./runconsumer.sh
 ```
 ## run with open source single redis container
  * stop docker if running from above
@@ -145,9 +153,11 @@ docker-compose up
 ```
 ```bash
 # edit port number in script to 6379
+export REDIS_URI=redis://localhost:12000
  ./runconsumerLUA.sh
  ```
 ```bash
+export REDIS_URI=redis://localhost:12000
 ./runproducerSingle.sh
 ```
 ## Run using AWS Cloudformation

cfndeploy.sh

@@ -1,4 +1,4 @@
 CAP1=CAPABILITY_NAMED_IAM
 CAP2=CAPABILITY_AUTO_EXPAND
 CAP3=CAPABILITY_IAM
-aws cloudformation deploy --template-file ./packaged-template.out --stack-name ${PROJECT_NAME} --capabilities $CAP1 $CAP2 $CAP3 --parameter-overrides ProjectName=$PROJECT_NAME S3BucketName=${S3_BUCKET} RedisDatabasePort=${REDISPORT} ClusterName=${CLUSTERNAME} HostedZoneName=${HOSTEDZONENAME} HostedZoneId=${HOSTEDZONEID} ClusterUserName=${CLUSTERUSERNAME} ClusterPassword=${CLUSTERPASSWORD} ClusterURL=${CLUSTERURL} LocalIp=${LOCAL_IP} S3Bucket=${S3_BUCKET} KeyPair=${KEY_PAIR} EC2InstanceType=${EC2_INSTANCE_TYPE} NumberInstances=${NUMBER_INSTANCES}
+aws cloudformation deploy --template-file ./packaged-template.out --stack-name ${PROJECT_NAME} --capabilities $CAP1 $CAP2 $CAP3 --parameter-overrides ProjectName=$PROJECT_NAME S3BucketName=${S3_BUCKET} RedisDatabasePort=${REDISPORT} ClusterName=${CLUSTERNAME} HostedZoneName=${HOSTEDZONENAME} HostedZoneId=${HOSTEDZONEID} ClusterUserName=${CLUSTERUSERNAME} ClusterPassword=${CLUSTERPASSWORD} ClusterURL=${CLUSTERURL} LocalIp=${LOCAL_IP} S3Bucket=${S3_BUCKET} KeyPair=${KEY_PAIR} EC2InstanceType=${EC2_INSTANCE_TYPE} NumberInstances=${NUMBER_INSTANCES} VpcCIDR=${VPC_CIDR} UseRedisElasticIP=${CREATE_ELASTIC_IPS}

runconsumerLUA.sh

@@ -1 +1 @@
-mvn exec:java -Dexec.mainClass="com.jphaugla.redis.streams.RedisStreams101ConsumerLUA" -Dexec.args="6379"
+mvn exec:java -Dexec.mainClass="com.jphaugla.redis.streams.RedisStreams101ConsumerLUA" 

setEnvironment.sh

@@ -32,3 +32,8 @@ export EC2_INSTANCE_TYPE=r4.2xlarge
 export KEY_PAIR=${PREFIX}_${REGION}
 # number of instances between 1 and 3
 export NUMBER_INSTANCES=2
+# VPC Cidr
+export VPC_CIDR=10.1.0.0/16
+#  Create Elastic IPs for fixed IPS on redis nodes
+export CREATE_ELASTIC_IPS=false
+export REDIS_URI=redis://redis-${REDISPORT}.${REGION}.${PREFIX}.demo-rlec.redislabs.com:${REDISPORT}

src/main/java/com/jphaugla/redis/streams/RedisStreams101ConsumerLUA.java

@@ -18,7 +18,7 @@ public class RedisStreams101ConsumerLUA {
     public final static String STREAMS_KEY = "weather_sensor:wind";
     public final static String HASH_INSIDE = ":hash:";
     //  to avoid cross slot error must use brackets so same hash slot
-    public final static String MESSAGE_PREFIX = "{wind:message:";
+    public final static String MESSAGE_PREFIX = "{weather_sensor:wind}:message";
     public final static String APPLICATION_NAME = "application_1";
 
     public static void main(String[] args) throws JsonProcessingException {
@@ -62,7 +62,7 @@ public static void main(String[] args) throws JsonProcessingException {
                     // syncCommands.xack(STREAMS_KEY, APPLICATION_NAME,  message.getId());
                     Map<String, String> body = message.getBody();
                     String messageId = body.get("message_id");
-                    String messageKey = MESSAGE_PREFIX + messageId + "}";
+                    String messageKey = MESSAGE_PREFIX + messageId;
                     System.out.println(messageKey);
                     String hashKey = messageKey + HASH_INSIDE + message.getId();
                     // System.out.println(hashKey);

templates/MasterWithRedis.yaml

@@ -146,6 +146,22 @@ Parameters:
       - i3.8xlarge
       - i3.16xlarge
     ConstraintDescription: Must be a valid EC2 instance type.
+  VpcCIDR:
+    Description: Enter the VPC CIDR range in the form x.x.x.x/16.
+    Type: String
+    MinLength: 9
+    MaxLength: 18
+    AllowedPattern: "(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})/(\\d{1,2})"
+    ConstraintDescription: Must be a valid CIDR range in the form x.x.x.x/16.
+    Default: 10.20.0.0/16
+  UseRedisElasticIP:
+    Description: Flag to fixed IPs for the redis nodes
+    Type: String
+    Default: false
+    AllowedValues:
+      - true
+      - false
+
 
 Conditions:
   Create-One: !Equals [ !Ref NumberInstances, 1 ]
@@ -154,6 +170,7 @@ Conditions:
   Do-Two: !Or
     - !Condition Create-Two
     - !Condition Create-Three
+  Create-ElasticIps: !Equals [ !Ref UseRedisElasticIP, true ]
 
 Resources:
   VPCStack:
@@ -164,15 +181,19 @@ Resources:
       Parameters:
         LocalIp:
           Ref: LocalIp
+        VpcCIDR:
+          Ref: VpcCIDR
     Metadata:
       'AWS::CloudFormation::Designer':
         id: 7aa57bac-380e-4523-a72b-ef898d0a0367
 
   IPAddressVM1:
     Type: AWS::EC2::EIP
+    Condition: Create-ElasticIps
 
   IPAssocVM1:
     Type: AWS::EC2::EIPAssociation
+    Condition: Create-ElasticIps
     Properties:
       InstanceId: !Ref RedisEC2VM1
       EIP: !Ref IPAddressVM1
@@ -181,7 +202,6 @@ Resources:
     Type:  AWS::EC2::Instance
     DependsOn: 
       - VPCStack
-      - IPAddressVM1
     Properties:
       ImageId: 
         Fn::FindInMap:
@@ -212,10 +232,12 @@ Resources:
 
   IPAddressVM2:
     Condition: Do-Two
+    Condition: Create-ElasticIps
     Type: AWS::EC2::EIP
 
   IPAssocVM2:
     Condition: Do-Two
+    Condition: Create-ElasticIps
     Type: AWS::EC2::EIPAssociation
     Properties:
       InstanceId: !Ref RedisEC2VM2
@@ -227,7 +249,6 @@ Resources:
     DependsOn: 
       - VPCStack
       - RedisEC2VM1
-      - IPAddressVM2
     Properties:
       ImageId: 
         Fn::FindInMap:
@@ -260,10 +281,12 @@ Resources:
 
   IPAddressVM3:
     Condition: Create-Three
+    Condition: Create-ElasticIps
     Type: AWS::EC2::EIP
 
   IPAssocVM3:
     Condition: Create-Three
+    Condition: Create-ElasticIps
     Type: AWS::EC2::EIPAssociation
     Properties:
       InstanceId: !Ref RedisEC2VM3
@@ -275,7 +298,6 @@ Resources:
     DependsOn: 
       - VPCStack
       - RedisEC2VM1
-      - IPAddressVM3
     Properties:
       ImageId: 
         Fn::FindInMap:
@@ -348,29 +370,29 @@ Resources:
   RedisARecord1:
     Type: AWS::Route53::RecordSet
     DependsOn: 
-      - IPAddressVM1
+      - RedisEC2VM1
     Properties:
       HostedZoneId: !Ref HostedZoneId
       Comment: DNS name for my instance
       Name: !Join ['',  ['node1.', !Ref ClusterName, ., !Ref HostedZoneName ] ]  
       Type: A
       TTL: 900
       ResourceRecords:
-        - !Ref IPAddressVM1
+        - !GetAtt RedisEC2VM1.PublicIp
 
   RedisARecord2:
     Condition: Do-Two
     Type: AWS::Route53::RecordSet
     DependsOn: 
-      - IPAddressVM2
+      - RedisEC2VM2
     Properties:
       HostedZoneId: !Ref HostedZoneId
       Comment: DNS name for my instance
       Name: !Join ['',  ['node2.', !Ref ClusterName, ., !Ref HostedZoneName ] ]  
       Type: A
       TTL: 900
       ResourceRecords:
-        - !Ref IPAddressVM2
+        - !GetAtt RedisEC2VM1.PublicIp
 
   RedisARecord3:
     Condition: Create-Three
@@ -384,7 +406,7 @@ Resources:
       Type: A
       TTL: 900
       ResourceRecords:
-        - !Ref IPAddressVM3
+        - !GetAtt RedisEC2VM1.PublicIp
 
   RedisNSRecord:
     Type: AWS::Route53::RecordSet

templates/VPCStack.yaml

@@ -5,15 +5,32 @@ Parameters:
   LocalIp:
     Description:  Local IP address to limit workgroup opening
     Type: String
+  VpcCIDR:
+    Description: Enter the VPC CIDR range in the form x.x.x.x/16.
+    Type: String
+    MinLength: 9
+    MaxLength: 18
+    AllowedPattern: "(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})/(\\d{1,2})"
+    ConstraintDescription: Must be a valid CIDR range in the form x.x.x.x/16.
+    Default: 10.20.0.0/16
 
 Resources:
   VPC:
     Type: 'AWS::EC2::VPC'
     Properties:
-      CidrBlock: '10.0.0.0/16'
+      CidrBlock: 
+        Ref: VpcCIDR
       EnableDnsSupport: true
       EnableDnsHostnames: true
       InstanceTenancy: default
+      Tags:
+      - Key: Name
+        Value:
+          Fn::Join:
+          - "-"
+          - - Ref: AWS::StackName
+            - Vpc
+
 
   InternetGateway:
     Type: 'AWS::EC2::InternetGateway'
@@ -28,9 +45,16 @@ Resources:
     Type: 'AWS::EC2::Subnet'
     Properties:
       AvailabilityZone: !Select [0, !GetAZs '']
-      CidrBlock: '10.0.0.1/20'
+      CidrBlock: !Select [ 0, !Cidr [ !GetAtt VPC.CidrBlock, 3, 8 ]]
       MapPublicIpOnLaunch: true
       VpcId: !Ref VPC
+      Tags:
+      - Key: Name
+        Value:
+          Fn::Join:
+          - "-"
+          - - Ref: AWS::StackName
+            - PublicSubnet1
 
   RouteTablePublicSubnet1:
     Type: 'AWS::EC2::RouteTable'