refactor TagList to be more intuitive

Previous behavior was based on some tests to reduce the
cost of generating the dynamic ids in high volume paths
compared to servo. However, it led to a number of surprising
and confusing issues for users. With this change we want
to clear up the behavior for users and we can optimize
as needed in the future.

It will now sort by the tag key and remove duplicates on
update. The goal is to have more intuitive behavior for
the user so they can rely on:

1. Any ids with the same set of tags will be equal.
2. Later additions will always override if the key already
   exists. So for `withTags(Iterable)` if there were
   entries with duplicate keys the iteration order would
   determine which one ended up getting used.

spectator-api/src/main/java/com/netflix/spectator/api/DefaultId.java

@@ -48,28 +48,30 @@ public DefaultId(String name) {
   }
 
   @Override public DefaultId withTag(Tag tag) {
-    return new DefaultId(name, new TagList(tag.key(), tag.value(), tags));
+    return new DefaultId(name, tags == TagList.EMPTY ? new TagList(tag.key(), tag.value()) : tags.mergeTag(tag));
   }
 
   @Override public DefaultId withTag(String key, String value) {
-    return new DefaultId(name, new TagList(key, value, tags));
+    TagList tag = new TagList(key, value);
+    return new DefaultId(name, tags == TagList.EMPTY ? tag : tags.mergeTag(tag));
   }
 
   @Override public DefaultId withTags(Iterable<Tag> ts) {
-    TagList tmp = (tags == TagList.EMPTY) ? TagList.create(ts) : tags.prepend(ts);
+    TagList tmp = (tags == TagList.EMPTY) ? TagList.create(ts) : tags.mergeList(ts);
     return new DefaultId(name, tmp);
   }
 
   @Override public DefaultId withTags(Map<String, String> ts) {
-    return withTags(TagList.create(ts));
+    TagList tmp = (tags == TagList.EMPTY) ? TagList.create(ts) : tags.mergeMap(ts);
+    return new DefaultId(name, tmp);
   }
 
   /**
    * Returns a new id with the tag list sorted by key and with no duplicate keys.  This is equivalent to
    * {@code rollup(Collections.<String>emptySet(), false)}.
    */
   DefaultId normalize() {
-    return rollup(Collections.<String>emptySet(), false);
+    return rollup(Collections.emptySet(), false);
   }
 
   /**
@@ -115,8 +117,8 @@ DefaultId rollup(Set<String> keys, boolean keep) {
   @Override public String toString() {
     StringBuilder buf = new StringBuilder();
     buf.append(name);
-    for (Tag t : tags()) {
-      buf.append(":").append(t.key()).append("=").append(t.value());
+    if (tags != TagList.EMPTY) {
+      buf.append(':').append(tags);
     }
     return buf.toString();
   }

spectator-api/src/main/java/com/netflix/spectator/api/TagList.java

@@ -17,15 +17,80 @@
 
 import com.netflix.spectator.impl.Preconditions;
 
+import java.util.Comparator;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.TreeMap;
 
 /**
- * A tag list implemented as a singly linked list. Doesn't automatically dedup keys but supports
- * a cheap prepend at the call site to allow for inexpensive dynamic ids.
+ * A tag list implemented as a singly linked list. The contents of the list are maintained in
+ * sorted order by key with no duplicates.
  */
 final class TagList implements Iterable<Tag>, Tag {
+  /**
+   * Utility class for sorting and deduplicating lists of tags.
+   */
+  private static class TagSorterAndDeduplicator {
+    private static final Comparator<String> REVERSE_STRING_COMPARATOR =
+        (String left, String right) -> right.compareTo(left);
+
+    /** Map used to sort and deduplicate the presented tags. */
+    private final Map<String, Tag> map;
+
+    /**
+     * Construct a new instance with no tags in it.
+     */
+    TagSorterAndDeduplicator() {
+      map  = new TreeMap<>(REVERSE_STRING_COMPARATOR);
+    }
+
+    /**
+     * Adds the specified tag to the collected tags. It will overwrite any existing value
+     * associated the key in the specified tag.
+     *
+     * @param tag
+     *     The tag to add to the collection.
+     */
+    void addTag(Tag tag) {
+      map.put(tag.key(), tag);
+    }
+
+    /**
+     * Adds the tags in the iterable to the collected tags. Any values associated with the tags in
+     * the iterable will overwrite any existing values with the same key that are already in the
+     * collection.
+     *
+     * @param tags
+     *     The set of tags to add.
+     */
+    void addTags(Iterable<Tag> tags) {
+      for (Tag t : tags) {
+        map.put(t.key(), t);
+      }
+    }
+
+    /**
+     * Adds the tags (key, value)-pairs to the collected tags. Any values associated with the tags
+     * in the map will overwrite any existing values with the same key that are already in the
+     * collection.
+     *
+     * @param tags
+     *     The set of tags to add.
+     */
+    void addTags(Map<String, String> tags) {
+      for (Map.Entry<String, String> t : tags.entrySet()) {
+        map.put(t.getKey(), new TagList(t.getKey(), t.getValue()));
+      }
+    }
+
+    /**
+     * Returns the sorted set of tags.
+     */
+    Iterable<Tag> sortedTags() {
+      return map.values();
+    }
+  }
 
   private final String key;
   private final String value;
@@ -40,9 +105,10 @@ final class TagList implements Iterable<Tag>, Tag {
   }
 
   /**
-   * Create a new instance with a new tag prepended to the list {@code next}.
+   * Create a new instance with a new tag prepended to the list {@code next}. Any entries in next
+   * should have keys that are lexicographically after the specified key.
    */
-  TagList(String key, String value, TagList next) {
+  private TagList(String key, String value, TagList next) {
     this.key = Preconditions.checkNotNull(key, "key");
     this.value = Preconditions.checkNotNull(value, "value");
     this.next = next;
@@ -101,20 +167,121 @@ public void remove() {
     return hc;
   }
 
+  @Override
+  public String toString() {
+    StringBuilder buf = new StringBuilder();
+    TagList cur = next;
+
+    buf.append(key).append('=').append(value);
+    while (cur != null) {
+      buf.append(":").append(cur.key()).append("=").append(cur.value());
+      cur = cur.next;
+    }
+    return buf.toString();
+  }
+
+  /**
+   * Produces a list with with the specified tag merged with the existing values in this list. If
+   * the key of the specified tag matches an existing list entry, then the value of the specified
+   * tag will replace the existing value.
+   *
+   * @param tag
+   *     Possibly null tag to merge into the list.
+   * @return
+   *     A tag list with merged values.
+   */
+  TagList mergeTag(Tag tag) {
+    if (tag == null) {
+      return this;
+    } else if (next == null) {
+      int comparison = key.compareTo(tag.key());
+
+      if (comparison == 0) { // Same key, so the specified value replaces the current value.
+        return new TagList(tag.key(), tag.value(), EMPTY);
+      } else if (comparison < 0) { // The key in this list is before the key in the specified list.
+        return new TagList(key, value, new TagList(tag.key(), tag.value(), EMPTY));
+      } else { // The key in this list is after the key in the specified list.
+        return new TagList(tag.key(), tag.value(), this);
+      }
+    } else {
+      // Is it possible to optimize this case so as to reuse the tail of the existing TagList?
+      TagSorterAndDeduplicator entries = new TagSorterAndDeduplicator();
+
+      entries.addTags(this);
+      entries.addTag(tag);
+
+      return createFromSortedTags(entries.sortedTags());
+    }
+  }
+
   /**
-   * Create a new list with the tags prepended to this list.
+   * Produces a list with the tags from this list merged with the tags in the specified list. For
+   * any keys present in both lists, the value from the specified list will replace the existing
+   * value.
    *
    * @param tags
-   *     A set of tags to prepend.
+   *     A set of tags to merge.
    * @return
-   *     New tag list with the tags prepended.
+   *     A tag list with the merged values. Based on the inputs the result may be this, tags, or
+   *     a new object.
    */
-  TagList prepend(Iterable<Tag> tags) {
-    TagList head = this;
-    for (Tag t : tags) {
-      head = new TagList(t.key(), t.value(), head);
+  TagList mergeList(Iterable<Tag> tags) {
+    if (tags == null) {
+      return this;
+    }
+
+    Iterator<Tag> iter = tags.iterator();
+
+    if (iter.hasNext()) {
+      Tag firstTag = iter.next();
+
+      if (iter.hasNext()) {
+        // Iterator has multiple entries so we need to sort them and remove any duplicates.
+        TagSorterAndDeduplicator entries = new TagSorterAndDeduplicator();
+
+        entries.addTags(this);
+        entries.addTags(tags);
+
+        return createFromSortedTags(entries.sortedTags());
+      } else {
+        // Single entry iterator.
+        return mergeTag(firstTag);
+      }
+    } else {
+      // Empty iterator
+      return this;
+    }
+  }
+
+  /**
+   * Produces a list with the tags from this list merged with the tags in the specified list. For
+   * any keys present in both lists, the value from the specified list will replace the existing
+   * value.
+   *
+   * @param tags
+   *     A set of tags to merge.
+   * @return
+   *     A tag list with the merged values. Based on the inputs the result may be this or a new
+   *     object.
+   */
+  TagList mergeMap(Map<String, String> tags) {
+    if (tags == null || tags.isEmpty()) {
+      return this;
+    }
+
+    if (tags.size() == 1) {
+      Map.Entry<String, String> entry = tags.entrySet().iterator().next();
+
+      return mergeTag(new TagList(entry.getKey(), entry.getValue(), EMPTY));
+    } else {
+        // Iterator has multiple entries so we need to sort them and remove any duplicates.
+        TagSorterAndDeduplicator entries = new TagSorterAndDeduplicator();
+
+        entries.addTags(this);
+        entries.addTags(tags);
+
+      return createFromSortedTags(entries.sortedTags());
     }
-    return head;
   }
 
   /**
@@ -129,11 +296,26 @@ static TagList create(Iterable<Tag> tags) {
     if (tags == EMPTY || tags instanceof TagList) {
       return (TagList) tags;
     } else {
-      TagList head = EMPTY;
-      for (Tag t : tags) {
-        head = new TagList(t.key(), t.value(), head);
+      Iterator<Tag> iter = tags.iterator();
+
+      if (iter.hasNext()) {
+        Tag firstTag = iter.next();
+
+        if (iter.hasNext()) {
+          // Iterator has multiple entries so we need to sort them and remove any duplicates.
+          TagSorterAndDeduplicator entries = new TagSorterAndDeduplicator();
+
+          entries.addTags(tags);
+
+          return createFromSortedTags(entries.sortedTags());
+        } else {
+          // Single entry iterator.
+          return new TagList(firstTag.key(), firstTag.value(), EMPTY);
+        }
+      } else {
+        // Empty iterator
+        return EMPTY;
       }
-      return head;
     }
   }
 
@@ -147,8 +329,37 @@ static TagList create(Iterable<Tag> tags) {
    */
   static TagList create(Map<String, String> tags) {
     TagList head = EMPTY;
-    for (Map.Entry<String, String> t : tags.entrySet()) {
-      head = new TagList(t.getKey(), t.getValue(), head);
+
+    if (tags.size() >= 2) {
+      TagSorterAndDeduplicator entries = new TagSorterAndDeduplicator();
+
+      for (Map.Entry<String, String> t : tags.entrySet()) {
+        entries.addTag(new TagList(t.getKey(), t.getValue()));
+      }
+      head = createFromSortedTags(entries.sortedTags());
+    } else {
+      for (Map.Entry<String, String> t : tags.entrySet()) {
+        head = new TagList(t.getKey(), t.getValue(), head);
+      }
+    }
+
+    return head;
+  }
+
+  /**
+   * Create a tag list from a sorted list with no duplicates. The TagList is created with the
+   * entries in the reverse order of the entries in the provided argument.
+   *
+   * @param sortedTags
+   *     The sorted collection of tags to use to create the list
+   * @return
+   *     The newly constructed tag list or {@code EMPTY} if the iterable contains no entries
+   */
+  private static TagList createFromSortedTags(Iterable<Tag> sortedTags) {
+    TagList head = EMPTY;
+
+    for (Tag t : sortedTags) {
+      head = new TagList(t.key(), t.value(), head);
     }
     return head;
   }

spectator-api/src/main/java/com/netflix/spectator/api/Utils.java

@@ -249,9 +249,9 @@ static Iterable<Tag> toIterable(String[] tags) {
     if (tags.length % 2 == 1) {
       throw new IllegalArgumentException("size must be even, it is a set of key=value pairs");
     }
-    TagList ts = TagList.EMPTY;
+    ArrayList<Tag> ts = new ArrayList<>(tags.length);
     for (int i = 0; i < tags.length; i += 2) {
-      ts = new TagList(tags[i], tags[i + 1], ts);
+      ts.add(new TagList(tags[i], tags[i + 1]));
     }
     return ts;
   }

spectator-api/src/test/java/com/netflix/spectator/api/CompatibilityTest.java

@@ -37,24 +37,24 @@ public class CompatibilityTest {
     EXPECTED.add("Measurement(collection-size,1234567890,8.0)");
     EXPECTED.add("Measurement(counter,1234567890,127.0)");
     EXPECTED.add("Measurement(counter:a=b,1234567890,381.0)");
+    EXPECTED.add("Measurement(dist:a=b:statistic=count,1234567890,15.0)");
+    EXPECTED.add("Measurement(dist:a=b:statistic=totalAmount,1234567890,504.0)");
     EXPECTED.add("Measurement(dist:statistic=count,1234567890,5.0)");
-    EXPECTED.add("Measurement(dist:statistic=count:a=b,1234567890,15.0)");
     EXPECTED.add("Measurement(dist:statistic=totalAmount,1234567890,168.0)");
-    EXPECTED.add("Measurement(dist:statistic=totalAmount:a=b,1234567890,504.0)");
     EXPECTED.add("Measurement(gauge,1234567890,49.0)");
     EXPECTED.add("Measurement(gauge-age,1234567890,0.049)");
     EXPECTED.add("Measurement(gauge-function,1234567890,65.0)");
-    EXPECTED.add("Measurement(gauge:node=i-12345:asg=foo-dev-v001:cluster=foo-dev:app=foo,1234567890,7.0)");
+    EXPECTED.add("Measurement(gauge:app=foo:asg=foo-dev-v001:cluster=foo-dev:node=i-12345,1234567890,7.0)");
+    EXPECTED.add("Measurement(long-timer:a=b:statistic=activeTasks,1234567890,3.0)");
+    EXPECTED.add("Measurement(long-timer:a=b:statistic=duration,1234567890,15120.0)");
     EXPECTED.add("Measurement(long-timer:statistic=activeTasks,1234567890,1.0)");
-    EXPECTED.add("Measurement(long-timer:statistic=activeTasks:a=b,1234567890,3.0)");
     EXPECTED.add("Measurement(long-timer:statistic=duration,1234567890,10080.0)");
-    EXPECTED.add("Measurement(long-timer:statistic=duration:a=b,1234567890,15120.0)");
     EXPECTED.add("Measurement(map-size,1234567890,8.0)");
     EXPECTED.add("Measurement(method-value,1234567890,22.0)");
+    EXPECTED.add("Measurement(timer:a=b:statistic=count,1234567890,24.0)");
+    EXPECTED.add("Measurement(timer:a=b:statistic=totalTime,1234567890,4.53852000126E14)");
     EXPECTED.add("Measurement(timer:statistic=count,1234567890,8.0)");
-    EXPECTED.add("Measurement(timer:statistic=count:a=b,1234567890,24.0)");
     EXPECTED.add("Measurement(timer:statistic=totalTime,1234567890,1.51284000042E14)");
-    EXPECTED.add("Measurement(timer:statistic=totalTime:a=b,1234567890,4.53852000126E14)");
   }
 
   @Test