jolynch

Every zone has a copy, and clients always read their local zone copy (via pooled memcached connections) first and fallback only once to another zone on miss. Key is staying in zone and memcached protocol plus super fast server latencies. It's been a little while since we measured, but memcached has a time to first byte of around 10us and then scales sublinearly with payload size [1]. Single zone latency is variable but generally between 150 and 250us roundtrip, cross AZ is terrible at up to a millisecond [2].

So you put 200us network with 30us response time and get about 250us average latency. Of course the P99 tail is closer to a millisecond and you have to do things like hedges to fight things like the hard coded eternity 200ms TCP packet retry timer ... But that's a whole other can of worms to talk about.

[1] https://github.com/Netflix-Skunkworks/service-capacity-model...

[2] https://jolynch.github.io/pdf/wlllb-apachecon-2022.pdf

(I work at Netflix on these Datastores)

EVCache definitely has some sharp edges and can be hard to use, which is one of the reasons we are putting it behind these gRPC abstractions like this Counter one or e.g. KeyValue [1] which offer CompletableFuture APIs with clean async and blocking modalities. We are also starting to add proper async APIs to EVCache itself e.g. getAsync [2] which the abstractions are using under-the-hood.

At the same time, EVCache is the cheapest (by about 10x in our experiments) caching solution with global replication [3] and cache warming [4] we are aware of. Every time we've tried alternatives like Redis or managed services they either fail to scale (e.g. cannot leverage flash storage effectively [5]) or cost waaay too much at our scale.

I absolutely agree though EVCache is probably the wrong choice for most folks - most folks aren't doing 100 million operations / second with 4-region full-active replication and applications that expect p50 client-side latency <500us. Similar I think to how most folks should probably start with PostgreSQL and not Cassandra.

[1] https://netflixtechblog.com/introducing-netflixs-key-value-d...

[2] https://github.com/Netflix/EVCache/blob/11b47ecb4e15234ca99c...

[3] https://www.infoq.com/articles/netflix-global-cache/

[4] https://netflixtechblog.medium.com/cache-warming-leveraging-...

[5] https://netflixtechblog.com/evolution-of-application-data-ca...

That's good to know - Spanner is even more impressive engineering!

I wish we had the staff and time to build and maintain something like Spanner. Different constraints lead to different solutions.

My experience has been that the talent density is the main difference. Netflix tackles huge problems with a small number of engineers. I think one angle of complexity you may be missing is efficiency - both in engineering cost and infrastructure cost.

Also YouTube has _excellent_ engineering (e.g. Vitess in the data space), and they are building atop an excellent infrastructure (e.g. Borg and the godly Google network). It's worth noting though that the whole Netflix infrastructure team is probably smaller than a small to medium satellite org at Google.

(post author)

It is indeed similar to DynamoDB as well as the original Cassandra Thrift API! This is intentional since those are both targeted backends and we need to be able to migrate customers between Cassandra Thrift, Cassandra CQL and DynamoDB. One of the most important things we use this abstraction for is seamless migration [1] as use cases and offerings evolve. Rather than think of KeyValue as the only database you ever need, think of it like your language's Map interface, and depending on the problem you are solving you need different implementations of that interface (different backing databases).

Graphs are indeed a challenge (and Relational is completely out of scope), but the high-scale Netflix graph abstraction is actually built atop KV just like a Graph library might be built on top of a language's built in Map type.

[1] https://www.youtube.com/watch?v=3bjnm1SXLlo

As a long time user and developer of databases, I would suggest isolation failures are not actually the source of most data related bugs. Most bugs I deal with are due to alternative failure modes like:

* We didn't think about how we would retry this operation when something fails or times out (idempotency)

* We didn't put the appropriate checksums in the right place (corruption)

* We didn't handle the load, often due to trying to provide stronger guarantees than the application needs, and went down causing lost operations (performance bottlenecks)

* We deployed bad software to the app or database, causing irreparable corruption that can't be fixed because we already purged the relevant commit/redo logs + snapshots.

I legitimately don't understand the calls for "SERIALIZABLE is the only valid isolation level" - I have not typically (ever that I can recall) seen at-scale production systems pay that cost for writes _and_ reads. Almost all applications I've seen (including banking/payment software) are fine with eventually consistent reads, as long as the staleness period is understood and reasonably bounded in time. Once you move past a single geographic datacenter, serializable writes become extremely expensive unless you can automatically home users to the appropriate leader datacenter, which most engineering teams can't guarantee.

The key is typically not isolation, it's modeling your application in an idempotent fashion that doesn't require isolation to be correct and keeping snapshots and those idempotent operation logs for a good few weeks at minimum. Maybe the Java analogy would be "if you can design it to not need locks, do that".

I don't fully agree for two reasons.

First, I am not sure the data on most in-use hardware (e.g. EC2 m5/c5/i3en etc ...) supports your conclusions. xxHash is faster than crypto hashes always and BLAKE3 single threaded is faster on every Intel machine I've come across in wide deployment. I hear similar arguments around CRC-32 and to be frank it just isn't true on most computers most people run things on.

Second, many languages don't properly use the hardware instructions and if they do they often don't use them correctly. For example, Java 8 has bog slow SHA-1, AES-GCM and MD5 implementations, and switching to Amazon Coretto Crypto Provider (which is just using proper native crypto) was able to speed SHA/MD5 up by 50% and AES-GCM by ~90% on a reasonably large deployment (although the JDK wasn't using proper hardware instructions for AES-GCM until Java 9 I think it is still slower even after that).

That being said, like I disclaimed at the top of the benchmark your particular hardware and your particular language matters a lot.

[1] https://github.com/corretto/amazon-corretto-crypto-provider/...

Agree with everything you say except that the post didn't mention non-cryptographic hashing algos that can be driven that hard. xxHash[1] (and especially XXH3) is almost always the fastest hashing choice, as it both is fast and has wide language support.

Sure there are some other fast ones out there like cityhash[2] but there aren't good Java/Python bindings I'm aware of and I wouldn't recommend using it in production given the lack of wide-spread use versus xxhash which is used by LZ4 internally and in databases all over the place.

[1] https://github.com/Cyan4973/xxHash [2] https://github.com/google/cityhash

>> or even just a high number of rounds of SHA-512

> Please god no :-)

Heh I didn't mean it as a recommendation per say, but I'm pretty sure Linux uses repeated SHA-512 on at least some common distros. At least on my Ubuntu Focal machine my /etc/shadow appears to be using SHA-512.

This is exactly the kind of stuff I wrote this post about. The first (exactly once) is actually just at-least-once with deduplication based on a counter. To reliably process the events, however, you need to make your downstream idempotent as well. Think of it like your event processor might fail, so even if you only receive the message "once" if you can fail processing it you still have to think about retries. In my opinion it would be explicitly better for the event system to provide "at least once" and intentionally duplicate events on occasion to test your processors ability to handle duplicates.

The second (lock) is actually a lease fwict, and writing code in the locked body that assumes it will be truly mutually exclusive is pretty dangerous (see the linked post from Martin [1] for why).

[1] https://martin.kleppmann.com/2016/02/08/how-to-do-distribute...

It conveys a false sense of correctness. Usually the system doing the processing has to use higher level or external methods of providing idempotency.

For example TCP implements "exactly once processing" by your definition but you probably still want Stripe to include idempotency keys in their charge API so you don't pay twice.

Great points, transactions are certainly useful in helping developers think about state transitions. I think some of the ~snark might come from my personal struggles with trying to convey why wrapping non idempotent state transitions in "BEGIN TRANSACTION ... COMMIT" doesn't immediately make the system reliable. I completely agree transactions make understanding the state transitions easier and that is valuable.

I do think CRDTs or idempotency/fencing tokens are also a valuable way to reason about state transitions, and they can provide much lower latency in a global distributed system.

Oh shoot I forgot Shibboleths have to remain secret, I have made a terrible mistake.

Thanks I'm glad you liked it! Your point on distributed transactions is very true, using CAS is what I meant by "transactionally advance a summary".

  For example, you could place a unique identifier on every count event and then roll 
  up those deltas in the background and transactionally advance a summary, either 
  preventing ingestion after some time delay or handling recounting.

Certainly transactions can help, but you still have to data model correctly for failure.

We ended up with a pretty different design this time around as we iterated on our highly available load balancing setup, so I figured that it was worth sharing out.

I think there are some useful lessons learned and tips/tricks in there even with the announcement that HAProxy will be supporting hitless reloads soon, but I look forward to the feedback!

Disclaimer: I help maintain Synapse.

I highly recommend this tool. Yelp has used it in production for years to manage a fairly large PaaS (hundreds of services, thousands of containers, constant churn); it's proven quite flexible and resilient.

Synapse is available on github [0], and we've open sourced our automation used to create a highly available service router using Synapse as well [1][2].

[0] https://github.com/airbnb/synapse [1] https://github.com/Yelp/synapse-tools/tree/master/src/synaps... [2] http://paasta.readthedocs.io/en/latest/yelpsoa_configs.html#...

Both NGINX and HAProxy will hang around for as long as the connection is open (up to the timeout). It's actually quite an issue when you're rapidly reloading either proxy (you can run out of memory reasonably easily), but most services that have long lived TCP connections also handle resets reasonably well so you can typically just kill the old proxies and it'll be ok.

I'm so excited about this. We just finished rolling out a new seamless strategy involving pairing NGINX with HAProxy which I am almost done with the blog post for, but I envision this making our solution even simpler in the future when it hits stable branches.

Absolutely awesome.

We've really enjoyed templating terraform. It solves a lot of the problems you're mentioning, and we can write the complex logic we need without interacting _directly_ with cloud APIs...

Compilers are like, useful.

When HAProxy reloads, it has to re-bind the sockets that it was listening on. Due to a bug in how Linux implements listen port sharing (SO_REUSEPORT), new incoming connections can get dropped for a very brief (~1ms) period while HAProxy reloads. The Github article and the linked Yelp article both go into detail on this.

The tldr is that in Linux right now there is no way to gracefully drain connections from a listening socket. Many programs work around this by passing the socket file descriptor from the old process to the new process, which is, for example, how nginx works.

Funny enough we're doing something pretty similar for our external lb tier but instead of a custom proxy we're just using nginx proxying back to HAProxy. Your solution is really nifty though because there aren't two accepts going on.

In the thing that we're trying nginx terminates TCP (and SSL if needed), proxies back to a unix socket where HAProxy is listening. This gives us the best of both worlds: we can reload HAProxy all day long and socket listeners are hitless, and we rarely need to reload the nginx config which comes out of the box with hitless TCP sockets. But it's not quite as cool as this solution because HAProxy and nginx both have to accept.

We haven't decided that we're going to do our internal load balancing this way yet because the listening ports change so often, but if we do I'll probably be working on https://github.com/airbnb/synapse/pull/203 to do it. The PR makes it so that Synapse can manage both nginx and HAProxy simultaneously (to deal with the extra system complexity of an additional proxy). Or we'll try this cool thing out :-P