]> git.saurik.com Git - redis.git/commitdiff
Merge remote branch 'pietern/unstable-zset' into unstable
authorantirez <antirez@gmail.com>
Thu, 14 Apr 2011 11:31:14 +0000 (13:31 +0200)
committerantirez <antirez@gmail.com>
Thu, 14 Apr 2011 11:31:14 +0000 (13:31 +0200)
TODO
src/Makefile
src/asciilogo.h [new file with mode: 0644]
src/cluster.c
src/config.c
src/redis-trib.rb [new file with mode: 0755]
src/redis.c
src/redis.h
src/sds.c
src/sds.h

diff --git a/TODO b/TODO
index e7943307239a4a7e6ffb7e7f3ede3246ed76e418..8f0a5db204cffb57630b16ff9b8824401ac2bb2e 100644 (file)
--- a/TODO
+++ b/TODO
@@ -9,19 +9,13 @@ WARNING: are you a possible Redis contributor?
          us, and *how* exactly this can be implemented to have good changes
          of a merge. Otherwise it is probably wasted work! Thank you
 
-DISKSTORE TODO
-==============
 
-* Fix FLUSHALL/FLUSHDB: the queue of pending reads/writes should be handled.
-* Check that 00/00 and ff/ff exist at startup, otherwise exit with error.
-* Implement sync flush option, where data is written synchronously on disk when a command is executed.
-* Implement MULTI/EXEC as transaction abstract API to diskstore.c, with transaction_start, transaction_end, and a journal to recover.
-* Stop BGSAVE thread on shutdown and any other condition where the child is killed during normal bgsave.
-* Fix RANDOMKEY to really do something interesting
-* Fix DBSIZE to really do something interesting
-* Add a DEBUG command to check if an entry is or not in memory currently
+API CHANGES
+===========
 
-* dscache.c near 236, kobj = createStringObject... we could use static obj.
+* Turn commands into variadic versions when it makes sense, that is, when
+  the variable number of arguments represent values, and there is no conflict
+  with the return value of the command.
 
 APPEND ONLY FILE
 ================
@@ -36,6 +30,8 @@ OPTIMIZATIONS
 * SORT: Don't copy the list into a vector when BY argument is constant.
 * Write the hash table size of every db in the dump, so that Redis can resize the hash table just one time when loading a big DB.
 * Read-only mode for slaves.
+* Redis big lists as linked lists of small ziplists?
+  Possibly a simple heuristic that join near nodes when some node gets smaller than the low_level, and split it into two if gets bigger than high_level.
 
 REPORTING
 =========
@@ -58,4 +54,21 @@ KNOWN BUGS
     What happens if between 1 and 2 for some reason (system under huge load
     or alike) too many time passes? We should prevent expires while the
     AOF is loading.
+* #519: Slave may have expired keys that were never read in the master (so a DEL
+  is not sent in the replication channel) but are already expired since
+  a lot of time. Maybe after a given delay that is undoubltly greater than
+  the replication link latency we should expire this key on the slave on
+  access?
 
+DISKSTORE TODO
+==============
+
+* Fix FLUSHALL/FLUSHDB: the queue of pending reads/writes should be handled.
+* Check that 00/00 and ff/ff exist at startup, otherwise exit with error.
+* Implement sync flush option, where data is written synchronously on disk when a command is executed.
+* Implement MULTI/EXEC as transaction abstract API to diskstore.c, with transaction_start, transaction_end, and a journal to recover.
+* Stop BGSAVE thread on shutdown and any other condition where the child is killed during normal bgsave.
+* Fix RANDOMKEY to really do something interesting
+* Fix DBSIZE to really do something interesting
+* Add a DEBUG command to check if an entry is or not in memory currently
+* dscache.c near 236, kobj = createStringObject... we could use static obj.
index f99be4b9c0dca2b7cd1157c48a90b14718c42e89..a566d4ea1141a5030ffffeb7f8e4efc4df836396 100644 (file)
@@ -83,7 +83,7 @@ redis-check-dump.o: redis-check-dump.c lzf.h
 redis-cli.o: redis-cli.c fmacros.h version.h ../deps/hiredis/hiredis.h \
   sds.h zmalloc.h ../deps/linenoise/linenoise.h help.h
 redis.o: redis.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \
-  zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h
+  zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h asciilogo.h
 release.o: release.c release.h
 replication.o: replication.c redis.h fmacros.h config.h ae.h sds.h dict.h \
   adlist.h zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h
diff --git a/src/asciilogo.h b/src/asciilogo.h
new file mode 100644 (file)
index 0000000..878a107
--- /dev/null
@@ -0,0 +1,18 @@
+char *ascii_logo =
+"                _._                                                  \n"
+"           _.-``__ ''-._                                             \n"
+"      _.-``    `.  `_.  ''-._           Redis %s (%s/%d) %s bit\n"
+"  .-`` .-```.  ```\\/    _.,_ ''-._                                   \n"
+" (    '      ,       .-`  | `,    )     Running in %s mode\n"
+" |`-._`-...-` __...-.``-._|'` _.-'|     Port: %d\n"
+" |    `-._   `._    /     _.-'    |     PID: %ld\n"
+"  `-._    `-._  `-./  _.-'    _.-'                                   \n"
+" |`-._`-._    `-.__.-'    _.-'_.-'|                                  \n"
+" |    `-._`-._        _.-'_.-'    |           http://redis.io        \n"
+"  `-._    `-._`-.__.-'_.-'    _.-'                                   \n"
+" |`-._`-._    `-.__.-'    _.-'_.-'|                                  \n"
+" |    `-._`-._        _.-'_.-'    |                                  \n"
+"  `-._    `-._`-.__.-'_.-'    _.-'                                   \n"
+"      `-._    `-.__.-'    _.-'                                       \n"
+"          `-._        _.-'                                           \n"
+"              `-.__.-'                                               \n\n";
index 783c658d43681d2d731e631c6f3bc7e6fa1b151a..d7e76ed437bed95708ea525c92e6ed3ade37917d 100644 (file)
@@ -11,6 +11,9 @@ void clusterSendFail(char *nodename);
 void clusterUpdateState(void);
 int clusterNodeGetSlotBit(clusterNode *n, int slot);
 sds clusterGenNodesDescription(void);
+clusterNode *clusterLookupNode(char *name);
+int clusterNodeAddSlave(clusterNode *master, clusterNode *slave);
+int clusterAddSlot(clusterNode *n, int slot);
 
 /* -----------------------------------------------------------------------------
  * Initialization
@@ -34,13 +37,104 @@ void clusterGetRandomName(char *p) {
 
 int clusterLoadConfig(char *filename) {
     FILE *fp = fopen(filename,"r");
+    char *line;
+    int maxline, j;
    
-    return REDIS_ERR;
     if (fp == NULL) return REDIS_ERR;
+
+    /* Parse the file. Note that single liens of the cluster config file can
+     * be really long as they include all the hash slots of the node.
+     * This means in the worst possible case REDIS_CLUSTER_SLOTS/2 integers.
+     * To simplify we allocate 1024+REDIS_CLUSTER_SLOTS*16 bytes per line. */
+    maxline = 1024+REDIS_CLUSTER_SLOTS*16;
+    line = zmalloc(maxline);
+    while(fgets(line,maxline,fp) != NULL) {
+        int argc;
+        sds *argv = sdssplitargs(line,&argc);
+        clusterNode *n, *master;
+        char *p, *s;
+
+        /* Create this node if it does not exist */
+        n = clusterLookupNode(argv[0]);
+        if (!n) {
+            n = createClusterNode(argv[0],0);
+            clusterAddNode(n);
+        }
+        /* Address and port */
+        if ((p = strchr(argv[1],':')) == NULL) goto fmterr;
+        *p = '\0';
+        memcpy(n->ip,argv[1],strlen(argv[1])+1);
+        n->port = atoi(p+1);
+
+        /* Parse flags */
+        p = s = argv[2];
+        while(p) {
+            p = strchr(s,',');
+            if (p) *p = '\0';
+            if (!strcasecmp(s,"myself")) {
+                redisAssert(server.cluster.myself == NULL);
+                server.cluster.myself = n;
+                n->flags |= REDIS_NODE_MYSELF;
+            } else if (!strcasecmp(s,"master")) {
+                n->flags |= REDIS_NODE_MASTER;
+            } else if (!strcasecmp(s,"slave")) {
+                n->flags |= REDIS_NODE_SLAVE;
+            } else if (!strcasecmp(s,"fail?")) {
+                n->flags |= REDIS_NODE_PFAIL;
+            } else if (!strcasecmp(s,"fail")) {
+                n->flags |= REDIS_NODE_FAIL;
+            } else if (!strcasecmp(s,"handshake")) {
+                n->flags |= REDIS_NODE_HANDSHAKE;
+            } else if (!strcasecmp(s,"noaddr")) {
+                n->flags |= REDIS_NODE_NOADDR;
+            } else if (!strcasecmp(s,"noflags")) {
+                /* nothing to do */
+            } else {
+                redisPanic("Unknown flag in redis cluster config file");
+            }
+            if (p) s = p+1;
+        }
+
+        /* Get master if any. Set the master and populate master's
+         * slave list. */
+        if (argv[3][0] != '-') {
+            master = clusterLookupNode(argv[3]);
+            if (!master) {
+                master = createClusterNode(argv[3],0);
+                clusterAddNode(master);
+            }
+            n->slaveof = master;
+            clusterNodeAddSlave(master,n);
+        }
+
+        /* Set ping sent / pong received timestamps */
+        if (atoi(argv[4])) n->ping_sent = time(NULL);
+        if (atoi(argv[5])) n->pong_received = time(NULL);
+
+        /* Populate hash slots served by this instance. */
+        for (j = 7; j < argc; j++) {
+            int start, stop;
+
+            if ((p = strchr(argv[j],'-')) != NULL) {
+                *p = '\0';
+                start = atoi(argv[j]);
+                stop = atoi(p+1);
+            } else {
+                start = stop = atoi(argv[j]);
+            }
+            while(start <= stop) clusterAddSlot(n, start++);
+        }
+
+        sdssplitargs_free(argv,argc);
+    }
+    zfree(line);
     fclose(fp);
 
+    /* Config sanity check */
+    redisAssert(server.cluster.myself != NULL);
     redisLog(REDIS_NOTICE,"Node configuration loaded, I'm %.40s",
         server.cluster.myself->name);
+    clusterUpdateState();
     return REDIS_OK;
 
 fmterr:
@@ -57,8 +151,8 @@ int clusterSaveConfig(void) {
     sds ci = clusterGenNodesDescription();
     int fd;
     
-    if ((fd = open(server.cluster.configfile,O_WRONLY|O_CREAT,0644)) == -1)
-        goto err;
+    if ((fd = open(server.cluster.configfile,O_WRONLY|O_CREAT|O_TRUNC,0644))
+        == -1) goto err;
     if (write(fd,ci,sdslen(ci)) != (ssize_t)sdslen(ci)) goto err;
     close(fd);
     sdsfree(ci);
@@ -79,7 +173,7 @@ void clusterSaveConfigOrDie(void) {
 void clusterInit(void) {
     int saveconf = 0;
 
-    server.cluster.myself = createClusterNode(NULL,REDIS_NODE_MYSELF);
+    server.cluster.myself = NULL;
     server.cluster.state = REDIS_CLUSTER_FAIL;
     server.cluster.nodes = dictCreate(&clusterNodesDictType,NULL);
     server.cluster.node_timeout = 15;
@@ -92,6 +186,7 @@ void clusterInit(void) {
     if (clusterLoadConfig(server.cluster.configfile) == REDIS_ERR) {
         /* No configuration found. We will just use the random name provided
          * by the createClusterNode() function. */
+        server.cluster.myself = createClusterNode(NULL,REDIS_NODE_MYSELF);
         redisLog(REDIS_NOTICE,"No cluster configuration found, I'm %.40s",
             server.cluster.myself->name);
         clusterAddNode(server.cluster.myself);
@@ -337,6 +432,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
                 /* Broadcast the failing node name to everybody */
                 clusterSendFail(node->name);
                 clusterUpdateState();
+                clusterSaveConfigOrDie();
             }
         } else {
             /* If it's not in NOADDR state and we don't have it, we
@@ -415,6 +511,7 @@ int clusterProcessPacket(clusterLink *link) {
 
     sender = clusterLookupNode(hdr->sender);
     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
+        int update_config = 0;
         redisLog(REDIS_DEBUG,"Ping packet received: %p", link->node);
 
         /* Add this node if it is new for us and the msg type is MEET.
@@ -428,6 +525,7 @@ int clusterProcessPacket(clusterLink *link) {
             nodeIp2String(node->ip,link);
             node->port = ntohs(hdr->port);
             clusterAddNode(node);
+            update_config = 1;
         }
 
         /* Get info from the gossip section */
@@ -435,8 +533,12 @@ int clusterProcessPacket(clusterLink *link) {
 
         /* Anyway reply with a PONG */
         clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
+
+        /* Update config if needed */
+        if (update_config) clusterSaveConfigOrDie();
     } else if (type == CLUSTERMSG_TYPE_PONG) {
-        int update = 0;
+        int update_state = 0;
+        int update_config = 0;
 
         redisLog(REDIS_DEBUG,"Pong packet received: %p", link->node);
         if (link->node) {
@@ -457,6 +559,7 @@ int clusterProcessPacket(clusterLink *link) {
                 redisLog(REDIS_DEBUG,"Handshake with node %.40s completed.",
                     link->node->name);
                 link->node->flags &= ~REDIS_NODE_HANDSHAKE;
+                update_config = 1;
             } else if (memcmp(link->node->name,hdr->sender,
                         REDIS_CLUSTER_NAMELEN) != 0)
             {
@@ -466,6 +569,7 @@ int clusterProcessPacket(clusterLink *link) {
                 redisLog(REDIS_DEBUG,"PONG contains mismatching sender ID");
                 link->node->flags |= REDIS_NODE_NOADDR;
                 freeClusterLink(link);
+                update_config = 1;
                 /* FIXME: remove this node if we already have it.
                  *
                  * If we already have it but the IP is different, use
@@ -511,7 +615,7 @@ int clusterProcessPacket(clusterLink *link) {
                             server.cluster.slots[j]->flags & REDIS_NODE_FAIL)
                         {
                             server.cluster.slots[j] = sender;
-                            update = 1;
+                            update_state = update_config = 1;
                         }
                     }
                 }
@@ -522,18 +626,21 @@ int clusterProcessPacket(clusterLink *link) {
         clusterProcessGossipSection(hdr,link);
 
         /* Update the cluster state if needed */
-        if (update) clusterUpdateState();
+        if (update_state) clusterUpdateState();
+        if (update_config) clusterSaveConfigOrDie();
     } else if (type == CLUSTERMSG_TYPE_FAIL && sender) {
         clusterNode *failing;
 
         failing = clusterLookupNode(hdr->data.fail.about.nodename);
-        if (failing && !(failing->flags & REDIS_NODE_FAIL)) {
+        if (failing && !(failing->flags & (REDIS_NODE_FAIL|REDIS_NODE_MYSELF)))
+        {
             redisLog(REDIS_NOTICE,
                 "FAIL message received from %.40s about %.40s",
                 hdr->sender, hdr->data.fail.about.nodename);
             failing->flags |= REDIS_NODE_FAIL;
             failing->flags &= ~REDIS_NODE_PFAIL;
             clusterUpdateState();
+            clusterSaveConfigOrDie();
         }
     } else {
         redisLog(REDIS_NOTICE,"Received unknown packet type: %d", type);
@@ -784,7 +891,7 @@ void clusterCron(void) {
              * normal PING packets. */
             node->flags &= ~REDIS_NODE_MEET;
 
-            redisLog(REDIS_NOTICE,"Connecting with Node %.40s at %s:%d\n", node->name, node->ip, node->port+REDIS_CLUSTER_PORT_INCR);
+            redisLog(REDIS_NOTICE,"Connecting with Node %.40s at %s:%d", node->name, node->ip, node->port+REDIS_CLUSTER_PORT_INCR);
         }
     }
     dictReleaseIterator(di);
@@ -814,22 +921,34 @@ void clusterCron(void) {
         int delay;
 
         if (node->flags &
-            (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE|
-             REDIS_NODE_FAIL)) continue;
+            (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE))
+                continue;
         /* Check only if we already sent a ping and did not received
          * a reply yet. */
         if (node->ping_sent == 0 ||
             node->ping_sent <= node->pong_received) continue;
 
         delay = time(NULL) - node->pong_received;
-        if (node->flags & REDIS_NODE_PFAIL) {
+        if (delay < server.cluster.node_timeout) {
             /* The PFAIL condition can be reversed without external
              * help if it is not transitive (that is, if it does not
-             * turn into a FAIL state). */
-            if (delay < server.cluster.node_timeout)
+             * turn into a FAIL state).
+             *
+             * The FAIL condition is also reversible if there are no slaves
+             * for this host, so no slave election should be in progress.
+             *
+             * TODO: consider all the implications of resurrecting a
+             * FAIL node. */
+            if (node->flags & REDIS_NODE_PFAIL) {
                 node->flags &= ~REDIS_NODE_PFAIL;
+            } else if (node->flags & REDIS_NODE_FAIL && !node->numslaves) {
+                node->flags &= ~REDIS_NODE_FAIL;
+                clusterUpdateState();
+            }
         } else {
-            if (delay >= server.cluster.node_timeout) {
+            /* Timeout reached. Set the noad se possibly failing if it is
+             * not already in this state. */
+            if (!(node->flags & (REDIS_NODE_PFAIL|REDIS_NODE_FAIL))) {
                 redisLog(REDIS_DEBUG,"*** NODE %.40s possibly failing",
                     node->name);
                 node->flags |= REDIS_NODE_PFAIL;
@@ -874,8 +993,7 @@ int clusterNodeGetSlotBit(clusterNode *n, int slot) {
  * an error and REDIS_ERR is returned. */
 int clusterAddSlot(clusterNode *n, int slot) {
     redisAssert(clusterNodeSetSlotBit(n,slot) == 0);
-    server.cluster.slots[slot] = server.cluster.myself;
-    printf("SLOT %d added to %.40s\n", slot, n->name);
+    server.cluster.slots[slot] = n;
     return REDIS_OK;
 }
 
@@ -967,8 +1085,8 @@ sds clusterGenNodesDescription(void) {
                 start = -1;
             }
         }
+        ci = sdscatlen(ci,"\n",1);
     }
-    ci = sdscatlen(ci,"\n",1);
     dictReleaseIterator(di);
     return ci;
 }
@@ -1047,6 +1165,7 @@ void clusterCommand(redisClient *c) {
         }
         zfree(slots);
         clusterUpdateState();
+        clusterSaveConfigOrDie();
         addReply(c,shared.ok);
     } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) {
         char *statestr[] = {"ok","fail","needhelp"};
@@ -1073,11 +1192,13 @@ void clusterCommand(redisClient *c) {
             "cluster_slots_ok:%d\r\n"
             "cluster_slots_pfail:%d\r\n"
             "cluster_slots_fail:%d\r\n"
+            "cluster_known_nodes:%lu\r\n"
             , statestr[server.cluster.state],
             slots_assigned,
             slots_ok,
             slots_pfail,
-            slots_fail
+            slots_fail,
+            dictSize(server.cluster.nodes)
         );
         addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
             (unsigned long)sdslen(info)));
index f6c6ad68fef93efe6868fc90427b7211b5012105..98fdb15d40d1b8b8407d46505a8dccbde6098880 100644 (file)
@@ -293,6 +293,9 @@ void loadServerConfig(char *filename) {
             if ((server.cluster_enabled = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"cluster-config-file") && argc == 2) {
+            zfree(server.cluster.configfile);
+            server.cluster.configfile = zstrdup(argv[1]);
         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }
diff --git a/src/redis-trib.rb b/src/redis-trib.rb
new file mode 100755 (executable)
index 0000000..31fc523
--- /dev/null
@@ -0,0 +1,258 @@
+#!/usr/bin/env ruby
+
+require 'rubygems'
+require 'redis'
+
+ClusterHashSlots = 4096
+
+def xputs(s)
+    printf s
+    STDOUT.flush
+end
+
+class ClusterNode
+    def initialize(addr)
+        s = addr.split(":")
+        if s.length != 2
+            puts "Invalid node name #{addr}"
+            exit 1
+        end
+        @r = nil
+        @host = s[0]
+        @port = s[1]
+        @slots = {}
+        @dirty = false
+    end
+
+    def to_s
+        "#{@host}:#{@port}"
+    end
+
+    def connect(o={})
+        xputs "Connecting to node #{self}: "
+        begin
+            @r = Redis.new(:host => @host, :port => @port)
+            @r.ping
+        rescue
+            puts "ERROR"
+            puts "Sorry, can't connect to node #{self}"
+            exit 1 if o[:abort]
+            @r = nil
+        end
+        puts "OK"
+    end
+
+    def assert_cluster
+        info = @r.info
+        if !info["cluster_enabled"] || info["cluster_enabled"].to_i == 0
+            puts "Error: Node #{self} is not configured as a cluster node."
+            exit 1
+        end
+    end
+
+    def assert_empty
+        if !(@r.cluster("info").split("\r\n").index("cluster_known_nodes:1")) ||
+            (@r.info['db0'])
+            puts "Error: Node #{self} is not empty. Either the node already knows other nodes (check with nodes-info) or contains some key in database 0."
+            exit 1
+        end
+    end
+
+    def add_slots(slots)
+        slots.each{|s|
+            @slots[s] = :new
+        }
+        @dirty = true
+    end
+
+    def flush_node_config
+        return if !@dirty
+        new = []
+        @slots.each{|s,val|
+            if val == :new
+                new << s
+                @slots[s] = true
+            end
+        }
+        @r.cluster("addslots",*new)
+        @dirty = false
+    end
+
+    def info_string
+        # We want to display the hash slots assigned to this node
+        # as ranges, like in: "1-5,8-9,20-25,30"
+        #
+        # Note: this could be easily written without side effects,
+        # we use 'slots' just to split the computation into steps.
+        
+        # First step: we want an increasing array of integers
+        # for instance: [1,2,3,4,5,8,9,20,21,22,23,24,25,30]
+        slots = @slots.keys.sort
+
+        # As we want to aggregate adiacent slots we convert all the
+        # slot integers into ranges (with just one element)
+        # So we have something like [1..1,2..2, ... and so forth.
+        slots.map!{|x| x..x}
+
+        # Finally we group ranges with adiacent elements.
+        slots = slots.reduce([]) {|a,b|
+            if !a.empty? && b.first == (a[-1].last)+1
+                a[0..-2] + [(a[-1].first)..(b.last)]
+            else
+                a + [b]
+            end
+        }
+
+        # Now our task is easy, we just convert ranges with just one
+        # element into a number, and a real range into a start-end format.
+        # Finally we join the array using the comma as separator.
+        slots = slots.map{|x|
+            x.count == 1 ? x.first.to_s : "#{x.first}-#{x.last}"
+        }.join(",")
+
+        "#{self.to_s.ljust(25)} slots:#{slots}"
+    end
+
+    def info
+        {
+            :host => @host,
+            :port => @port,
+            :slots => @slots,
+            :dirty => @dirty
+        }
+    end
+    
+    def is_dirty?
+        @dirty
+    end
+
+    def r
+        @r
+    end
+end
+
+class RedisTrib
+    def initialize
+        @nodes = []
+    end
+
+    def check_arity(req_args, num_args)
+        if ((req_args > 0 and num_args != req_args) ||
+           (req_args < 0 and num_args < req_args.abs))
+           puts "Wrong number of arguments for specified sub command"
+           exit 1
+        end
+    end
+
+    def add_node(node)
+        @nodes << node
+    end
+
+    def check_cluster
+        puts "Performing Cluster Check (using node #{@nodes[0]})"
+        show_nodes
+    end
+
+    def alloc_slots
+        slots_per_node = ClusterHashSlots/@nodes.length
+        i = 0
+        @nodes.each{|n|
+            first = i*slots_per_node
+            last = first+slots_per_node-1
+            last = ClusterHashSlots-1 if i == @nodes.length-1
+            n.add_slots first..last
+            i += 1
+        }
+    end
+
+    def flush_nodes_config
+        @nodes.each{|n|
+            n.flush_node_config
+        }
+    end
+
+    def show_nodes
+        @nodes.each{|n|
+            puts n.info_string
+        }
+    end
+
+    def join_cluster
+        # We use a brute force approach to make sure the node will meet
+        # each other, that is, sending CLUSTER MEET messages to all the nodes
+        # about the very same node.
+        # Thanks to gossip this information should propagate across all the
+        # cluster in a matter of seconds.
+        first = false
+        @nodes.each{|n|
+            if !first then first = n.info; next; end # Skip the first node
+            n.r.cluster("meet",first[:host],first[:port])
+        }
+    end
+
+    def yes_or_die(msg)
+        print "#{msg} (type 'yes' to accept): "
+        STDOUT.flush
+        if !(STDIN.gets.chomp.downcase == "yes")
+            puts "Aborting..."
+            exit 1
+        end
+    end
+
+    # redis-trib subcommands implementations
+
+    def check_cluster_cmd
+        node = ClusterNode.new(ARGV[1])
+        node.connect(:abort => true)
+        node.assert_cluster
+        add_node(node)
+        check_cluster
+    end
+
+    def create_cluster_cmd
+        puts "Creating cluster"
+        ARGV[1..-1].each{|n|
+            node = ClusterNode.new(n)
+            node.connect(:abort => true)
+            node.assert_cluster
+            node.assert_empty
+            add_node(node)
+        }
+        puts "Performing hash slots allocation on #{@nodes.length} nodes..."
+        alloc_slots
+        show_nodes
+        yes_or_die "Can I set the above configuration?"
+        flush_nodes_config
+        puts "** Nodes configuration updated"
+        puts "** Sending CLUSTER MEET messages to join the cluster"
+        join_cluster
+        check_cluster
+    end
+end
+
+COMMANDS={
+    "create" => ["create_cluster_cmd", -2, "host1:port host2:port ... hostN:port"],
+    "check" =>  ["check_cluster_cmd", 2, "host:port"]
+}
+
+# Sanity check
+if ARGV.length == 0
+    puts "Usage: redis-trib <command> <arguments ...>"
+    puts
+    COMMANDS.each{|k,v|
+        puts "  #{k.ljust(20)} #{v[2]}"
+    }
+    puts
+    exit 1
+end
+
+rt = RedisTrib.new
+cmd_spec = COMMANDS[ARGV[0].downcase]
+if !cmd_spec
+    puts "Unknown redis-trib subcommand '#{ARGV[0]}'"
+    exit 1
+end
+rt.check_arity(cmd_spec[1],ARGV.length)
+
+# Dispatch
+rt.send(cmd_spec[0])
index 3d0f53788640ce7f1c5d52e2b3d3faadefadf52c..b0c3179e288762bebc98b930894b80b0f779b7a5 100644 (file)
@@ -205,14 +205,20 @@ void redisLogRaw(int level, const char *msg) {
     time_t now = time(NULL);
     FILE *fp;
     char buf[64];
+    int rawmode = (level & REDIS_LOG_RAW);
 
+    level &= 0xff; /* clear flags */
     if (level < server.verbosity) return;
 
     fp = (server.logfile == NULL) ? stdout : fopen(server.logfile,"a");
     if (!fp) return;
 
-    strftime(buf,sizeof(buf),"%d %b %H:%M:%S",localtime(&now));
-    fprintf(fp,"[%d] %s %c %s\n",(int)getpid(),buf,c[level],msg);
+    if (rawmode) {
+        fprintf(fp,"%s",msg);
+    } else {
+        strftime(buf,sizeof(buf),"%d %b %H:%M:%S",localtime(&now));
+        fprintf(fp,"[%d] %s %c %s\n",(int)getpid(),buf,c[level],msg);
+    }
     fflush(fp);
 
     if (server.logfile) fclose(fp);
@@ -227,7 +233,7 @@ void redisLog(int level, const char *fmt, ...) {
     va_list ap;
     char msg[REDIS_MAX_LOGMSG_LEN];
 
-    if (level < server.verbosity) return;
+    if ((level&0xff) < server.verbosity) return;
 
     va_start(ap, fmt);
     vsnprintf(msg, sizeof(msg), fmt, ap);
@@ -1500,6 +1506,15 @@ sds genRedisInfoString(char *section) {
         }
     }
 
+    /* Clusetr */
+    if (allsections || defsections || !strcasecmp(section,"cluster")) {
+        if (sections++) info = sdscat(info,"\r\n");
+        info = sdscatprintf(info,
+        "# Cluster\r\n"
+        "cluster_enabled:%d\r\n",
+        server.cluster_enabled);
+    }
+
     /* Key space */
     if (allsections || defsections || !strcasecmp(section,"keyspace")) {
         if (sections++) info = sdscat(info,"\r\n");
@@ -1708,6 +1723,23 @@ void usage() {
     exit(1);
 }
 
+void redisAsciiArt(void) {
+#include "asciilogo.h"
+    char *buf = zmalloc(1024*16);
+
+    snprintf(buf,1024*16,ascii_logo,
+        REDIS_VERSION,
+        redisGitSHA1(),
+        strtol(redisGitDirty(),NULL,10) > 0,
+        (sizeof(long) == 8) ? "64" : "32",
+        server.cluster_enabled ? "cluster" : "stand alone",
+        server.port,
+        (long) getpid()
+    );
+    redisLogRaw(REDIS_NOTICE|REDIS_LOG_RAW,buf);
+    zfree(buf);
+}
+
 int main(int argc, char **argv) {
     long long start;
 
@@ -1726,6 +1758,7 @@ int main(int argc, char **argv) {
     if (server.daemonize) daemonize();
     initServer();
     if (server.daemonize) createPidFile();
+    redisAsciiArt();
     redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
 #ifdef __linux__
     linuxOvercommitMemoryWarning();
index 32dcc359077de570d4fefa73b11901e6ec29d9f6..5506e36571e9f7ac219b2eb64d4e8ade7da11e34 100644 (file)
 #define REDIS_VERBOSE 1
 #define REDIS_NOTICE 2
 #define REDIS_WARNING 3
+#define REDIS_LOG_RAW (1<<10) /* Modifier to log without timestamp */
 
 /* Anti-warning macro... */
 #define REDIS_NOTUSED(V) ((void) V)
index 8c7540990e8154d0326f885e2a529da062acf21f..37c63b055a241880019c31720994e8f1fd125867 100644 (file)
--- a/src/sds.c
+++ b/src/sds.c
@@ -554,6 +554,13 @@ err:
     return NULL;
 }
 
+void sdssplitargs_free(sds *argv, int argc) {
+    int j;
+
+    for (j = 0 ;j < argc; j++) sdsfree(argv[j]);
+    zfree(argv);
+}
+
 #ifdef SDS_TEST_MAIN
 #include <stdio.h>
 #include "testhelp.h"
index 91a387821bb267b1084f09b340bd7aaef6f5e5d5..6a3178688f5040af7572deb72f628d507cabe4ee 100644 (file)
--- a/src/sds.h
+++ b/src/sds.h
@@ -74,5 +74,6 @@ void sdstoupper(sds s);
 sds sdsfromlonglong(long long value);
 sds sdscatrepr(sds s, char *p, size_t len);
 sds *sdssplitargs(char *line, int *argc);
+void sdssplitargs_free(sds *argv, int argc);
 
 #endif